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

Merge pull request #16 from travisturner/disco-config-noder-primaryid-remove-gossip

Browse source
This commit is contained in:
Antonio Navarro Perez 2021-02-04 10:45:23 +01:00 • committed by GitHub
commit 39eacfb9a3
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 10 additions and 427 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

312
cluster.go
View file

@ -923,92 +923,6 @@ func (c *cluster) markAsJoined() {
}
}
// needTopologyAgreement is unprotected.
func (c *cluster) needTopologyAgreement() bool {
return false
}
// haveTopologyAgreement is unprotected.
func (c *cluster) haveTopologyAgreement() bool {
if c.Static {
return true
}
return stringSlicesAreEqual(c.Topology.nodeIDs, c.nodeIDs())
}
// allNodesReady is unprotected.
func (c *cluster) allNodesReady() (ret bool) {
if c.Static {
return true
}
nodeStates, err := c.stator.NodeStates(context.TODO())
if err != nil {
c.logger.Printf("getting node states error: %v", err)
return false
}
for _, s := range nodeStates {
if s != disco.NodeStateStarted {
return false
}
}
return true
}
func (c *cluster) handleNodeAction(nodeAction nodeAction) error {
c.mu.Lock()
j, err := c.unprotectedGenerateResizeJob(nodeAction)
c.mu.Unlock()
if err != nil {
c.logger.Printf("generateResizeJob error: err=%s", err)
return errors.Wrap(err, "setting state")
}
// j.Run() runs in a goroutine because in the case where the
// job requires no action, it immediately writes to the j.result
// channel, which is not consumed until the code below.
var eg errgroup.Group
eg.Go(func() error {
return j.run()
})
// Wait for the resizeJob to finish or be aborted.
c.logger.Printf("wait for jobResult")
var jobResult string
select {
case <-c.closing:
return errors.New("cluster shut down during resize")
case jobResult = <-j.result:
}
// Make sure j.run() didn't return an error.
if eg.Wait() != nil {
return errors.Wrap(err, "running job")
}
c.logger.Printf("received jobResult: %s", jobResult)
switch jobResult {
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 {
c.mu.Lock()
defer c.mu.Unlock()
return c.removeNode(nodeAction.node.ID)
} else if j.action == resizeJobActionAdd {
c.mu.Lock()
defer c.mu.Unlock()
return c.addNode(nodeAction.node)
}
case resizeJobStateAborted:
if err := c.completeCurrentJob(resizeJobStateAborted); err != nil {
return errors.Wrap(err, "completing aborted job")
}
}
return nil
}
func (c *cluster) sendTo(node *topology.Node, m Message) error {
if err := c.broadcaster.SendTo(node, m); err != nil {
return errors.Wrap(err, "sending")
@ -1016,70 +930,6 @@ func (c *cluster) sendTo(node *topology.Node, m Message) error {
return nil
}
// listenForJoins handles cluster-resize events.
func (c *cluster) listenForJoins() {
c.wg.Add(1)
go func() {
defer c.wg.Done()
// 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.
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)
continue
}
continue
default:
}
// 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
}
continue
}
}
}()
}
// unprotectedGenerateResizeJob 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) unprotectedGenerateResizeJob(nodeAction nodeAction) (*resizeJob, error) {
c.logger.Printf("generateResizeJob: %v", nodeAction)
j, err := c.unprotectedGenerateResizeJobByAction(nodeAction)
if err != nil {
return nil, errors.Wrap(err, "generating job")
}
c.logger.Printf("generated resizeJob: %d", j.ID)
// Save job in jobs map for future reference.
c.jobs[j.ID] = j
// Set job as currentJob.
if c.currentJob != nil {
return nil, fmt.Errorf("there is currently a resize job running")
}
c.currentJob = j
return j, nil
}
// unprotectedGenerateResizeJobByAction 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
@ -1456,28 +1306,6 @@ func (j *resizeJob) setState(state string) {
j.mu.Unlock()
}
// run distributes ResizeInstructions.
func (j *resizeJob) run() error {
j.Logger.Printf("run resizeJob")
// Set job state to RUNNING.
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
return nil
}
j.Logger.Printf("distribute tasks for resizeJob")
err := j.distributeResizeInstructions()
if err != nil {
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 {
switch j.state {
@ -1498,25 +1326,6 @@ func (j *resizeJob) nodesArePending() bool {
return false
}
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.
for _, instr := range j.Instructions {
// Because the node may not be in the cluster yet, create
// a dummy node object to use in the SendTo() method.
node := &topology.Node{
ID: instr.Node.ID,
URI: instr.Node.URI,
GRPCURI: instr.Node.GRPCURI,
}
j.Logger.Printf("send resize instructions: %v", instr)
if err := j.Broadcaster.SendTo(node, instr); err != nil {
return errors.Wrap(err, "sending instruction")
}
}
return nil
}
type nodeIDs []string
func (n nodeIDs) Len() int { return len(n) }
@ -1819,127 +1628,6 @@ func (c *cluster) confirmNodeDown(uri pnet.URI) bool {
return true
}
// ReceiveEvent represents an implementation of EventHandler.
func (c *cluster) ReceiveEvent(e *NodeEvent) (err error) {
// Ignore events sent from this node.
if e.Node.ID == c.Node.ID {
return nil
}
switch e.Event {
case NodeJoin:
// Ignore the event if this is not the coordinator.
if !c.isCoordinator() {
return nil
}
return c.nodeJoin(e.Node)
case NodeLeave:
c.mu.Lock()
defer c.mu.Unlock()
if c.unprotectedIsCoordinator() {
c.logger.Printf("received node leave: %v", e.Node)
// if removeNodeBasicSorted succeeds, that means that the node was
// not already removed by a removeNode request. We treat this as the
// host being temporarily unavailable, and expect it to come back
// up.
if c.confirmNodeDown(e.Node.URI) {
if c.removeNodeBasicSorted(e.Node.ID) {
c.Topology.nodeStates[e.Node.ID] = nodeStateDown
// put the cluster into STARTING if we've lost a number of nodes
// equal to or greater than ReplicaN
}
} else {
c.logger.Printf("ignored received node leave: %v", e.Node)
}
}
case NodeUpdate:
c.logger.Printf("received node update event: id: %v, string: %v, uri: %v", e.Node.ID, e.Node.String(), e.Node.URI)
// NodeUpdate is intentionally not implemented.
}
return err
}
// nodeJoin should only be called by the coordinator.
func (c *cluster) nodeJoin(node *topology.Node) error {
c.abortAntiEntropy()
// Technically there is a race condition here which could
// allow the anti-entropy process to re-start (and acquire
// the lock) before this lock has time to succeed. In that
// case, the user would have to wait through an entire
// anti-entropy cycle. We decided it wasn't worth the
// complexity (of, for example, implementing this with
// channels) to avoid that rare case.
c.muAntiEntropy.Lock()
defer c.muAntiEntropy.Unlock()
c.mu.Lock()
defer c.mu.Unlock()
c.logger.Printf("node join event on coordinator, node: %s, id: %s", node.URI, node.ID)
if c.needTopologyAgreement() {
// A host that is not part of the topology can't be added to the STARTING cluster.
if !c.Topology.ContainsID(node.ID) {
err := fmt.Sprintf("host is not in topology: %s", node.ID)
c.logger.Printf("%v", err)
return errors.New(err)
}
if err := c.addNode(node); err != nil {
return errors.Wrap(err, "adding node for agreement")
}
// Only change to normal if there is no existing data. Otherwise,
// the coordinator needs to wait to receive READY messages (nodeStates)
// from remote nodes before setting the cluster to state NORMAL.
if ok, err := c.holder.HasData(); !ok && err == nil {
// If the result of the previous AddNode completed the joining of nodes
// in the topology, then change the state to NORMAL.
if c.haveTopologyAgreement() {
return nil
}
// This lets the remote node to proceed with opening its holder,
// instead of waiting in DOWN state because cluster is in STARTING state.
return c.sendTo(node, c.unprotectedStatus())
} else if err != nil {
return errors.Wrap(err, "checking if holder has data")
}
if c.haveTopologyAgreement() && c.allNodesReady() {
return nil
}
// Send the status to the remote node. This lets the remote node
// know that it can proceed with opening its Holder.
return c.sendTo(node, c.unprotectedStatus())
}
// 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 cnode := c.unprotectedNodeByID(node.ID); cnode != nil {
if cnode.URI != node.URI {
c.logger.Printf("node: %v changed URI from %s to %s", cnode.ID, cnode.URI, node.URI)
cnode.URI = node.URI
}
if cnode.GRPCURI != node.GRPCURI {
cnode.GRPCURI = node.GRPCURI
}
return nil
}
// 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 {
return errors.Wrap(err, "adding node")
}
return nil
} else if err != nil {
return errors.Wrap(err, "checking if holder has data2")
}
c.joiningLeavingNodes <- nodeAction{node, resizeJobActionAdd}
return nil
}
// nodeLeave initiates the removal of a node from the cluster.
func (c *cluster) nodeLeave(nodeID string) error {
c.abortAntiEntropy()

24
pilosa.go
View file

@ -181,30 +181,6 @@ func validateName(name string) error {
return nil
}
// stringSlicesAreEqual determines if two string slices are equal.
func stringSlicesAreEqual(a, b []string) bool {
if a == nil && b == nil {
return true
}
if a == nil || b == nil {
return false
}
if len(a) != len(b) {
return false
}
for i := range a {
if a[i] != b[i] {
return false
}
}
return true
}
func timestamp() int64 {
return time.Now().UnixNano()
}

26
server.go
View file

@ -73,9 +73,6 @@ type Server struct { // nolint: maligned
sharder disco.Sharder
schemator disco.Schemator
// TODO: this is VERY temporary!!!
Gossiper Gossiper
// External
systemInfo SystemInfo
gcNotifier GCNotifier
@ -527,10 +524,6 @@ func (s *Server) UpAndDown() error {
return nil
}
type Gossiper interface {
StartGossip() error
}
// Open opens and initializes the server.
func (s *Server) Open() error {
s.logger.Printf("open server. PID %v", os.Getpid())
@ -597,13 +590,6 @@ func (s *Server) Open() error {
return errors.Wrap(err, "setting up cluster")
}
// ---------- TODO: this is temporary
if s.Gossiper != nil {
if err := s.Gossiper.StartGossip(); err != nil {
return errors.Wrap(err, "starting gossip")
}
}
// Open Cluster management.
if err := s.cluster.waitForStarted(); err != nil {
return errors.Wrap(err, "opening Cluster")
@ -617,13 +603,6 @@ func (s *Server) Open() error {
s.holder.SnapshotQueue = s.snapshotQueue
s.holder.Activate()
// Listen for joining nodes.
// This needs to start after the Holder has opened so that nodes can join
// 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()
// if we joined existing cluster then broadcast "resize on add" message
// TODO
// if initState == disco.InitialClusterStateExisting {
@ -884,11 +863,6 @@ func (s *Server) receiveMessage(m Message) error {
}
case *RecalculateCaches:
s.holder.recalculateCaches()
case *NodeEvent:
err := s.cluster.ReceiveEvent(obj)
if err != nil {
return errors.Wrapf(err, "cluster receiving NodeEvent %v", obj)
}
case *NodeStatus:
s.handleRemoteStatus(obj)
case *TransactionMessage:

18
server/cluster_test.go
View file

@ -173,7 +173,7 @@ func TestClusterResize_AddNode(t *testing.T) {
m0 := test.MustRunCluster(t, 1).GetNode(0)
defer m0.Close()
seed := m0.GossipAddress()
seed := ""
// Create a client for each node.
client0 := m0.Client()
@ -219,7 +219,7 @@ func TestClusterResize_AddNode(t *testing.T) {
m0 := test.MustRunCluster(t, 1).GetNode(0)
defer m0.Close()
seed := m0.GossipAddress()
seed := ""
// Create a client for each node.
client0 := m0.Client()
@ -283,7 +283,7 @@ func TestClusterResize_AddNode(t *testing.T) {
m0 := test.MustRunCluster(t, 1).GetNode(0)
defer m0.Close()
seed := m0.GossipAddress()
seed := ""
// Create a client for each node.
client0 := m0.Client()
@ -345,7 +345,7 @@ func TestClusterResize_AddNode(t *testing.T) {
m0 := test.MustRunCluster(t, 1).GetNode(0)
defer m0.Close()
seed := m0.GossipAddress()
seed := ""
// Create a client for each node.
client0 := m0.Client()
@ -416,7 +416,7 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
m0 := test.MustRunCluster(t, 1).GetNode(0)
defer m0.Close()
seed := m0.GossipAddress()
seed := ""
// Create a client for each node.
client0 := m0.Client()
@ -468,7 +468,7 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
m0 := test.MustRunCluster(t, 1).GetNode(0)
defer m0.Close()
seed := m0.GossipAddress()
seed := ""
// Create a client for each node.
client0 := m0.Client()
@ -536,7 +536,7 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
m0 := test.MustRunCluster(t, 1).GetNode(0)
defer m0.Close()
seed := m0.GossipAddress()
seed := ""
// Create a client for each node.
client0 := m0.Client()
@ -604,7 +604,7 @@ func TestClusterResize_AddNodeConcurrentIndex(t *testing.T) {
m0 := test.MustRunCluster(t, 1).GetNode(0)
defer m0.Close()
seed := m0.GossipAddress()
seed := ""
// Create a client for each node.
client0 := m0.Client()
@ -672,7 +672,7 @@ func TestCluster_GossipMembership(t *testing.T) {
m0 := test.MustRunCluster(t, 1).GetNode(0)
defer m0.Close()
seed := m0.GossipAddress()
seed := ""
var eg errgroup.Group

7
server/server.go
View file

@ -151,10 +151,6 @@ func NewCommand(stdin io.Reader, stdout, stderr io.Writer, opts ...CommandOption
return c
}
func (m *Command) StartGossip() (err error) {
return m.setupNetworking()
}
// Start starts the pilosa server - it returns once the server is running.
func (m *Command) Start() (err error) {
// Seed random number generator
@ -166,9 +162,6 @@ func (m *Command) Start() (err error) {
return errors.Wrap(err, "setting up server")
}
// TODO: this is temporary.
m.Server.Gossiper = m
if runtime.GOOS == "linux" {
result, err := ioutil.ReadFile("/proc/sys/vm/max_map_count")
if err != nil {

7
test/pilosa.go
View file

@ -108,13 +108,6 @@ func RunCommand(t *testing.T) *Command {
return MustRunCluster(t, 1).GetNode(0)
}
// GossipAddress returns the address on which gossip is listening after a Main
// has been setup. Useful to pass as a seed to other nodes when creating and
// testing clusters.
func (m *Command) GossipAddress() string {
return m.GossipTransport().URI.String()
}
// Close closes the program and removes the underlying data directory.
func (m *Command) Close() error {
// leave the removing part to the test logic. Some tests are closing and opening again the command

2
translator_test.go
View file

@ -263,7 +263,7 @@ func TestTranslation_Reset(t *testing.T) {
if err := node0.SoftOpen(); err != nil {
t.Fatal(err)
}
gossipSeeds := []string{node0.GossipAddress()}
gossipSeeds := []string{}
node1.Config.Gossip.Seeds = gossipSeeds
if err := node1.SoftOpen(); err != nil {

41
utils_internal_test.go
View file

@ -211,40 +211,6 @@ func (t *ClusterCluster) clusterByID(id string) *cluster {
// addNode adds a node to the cluster and (potentially) starts a resize job.
func (t *ClusterCluster) addNode() error {
id := len(t.Clusters)
c, err := t.addCluster(id, false)
if err != nil {
return err
}
// Send NodeJoin event to coordinator.
if id > 0 {
coord := t.Clusters[0]
ev := &NodeEvent{
Event: NodeJoin,
Node: c.Node,
}
if err := coord.ReceiveEvent(ev); err != nil {
return err
}
state, err := coord.State()
if err != nil {
return err
}
// Wait for the AddNode job to finish.
if state != string(ClusterStateNormal) {
t.resizeDone = make(chan struct{})
t.mu.Lock()
t.resizing = true
t.mu.Unlock()
<-t.resizeDone
}
}
return nil
}
@ -338,13 +304,6 @@ func (t *ClusterCluster) Open() error {
return err
}
}
// Start the listener on the coordinator.
if len(t.Clusters) == 0 {
return nil
}
t.Clusters[0].listenForJoins()
return nil
}