From afc53e1163c969b5aa304e7cf776c69bb64314c7 Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 3 Feb 2021 23:31:38 -0600 Subject: [PATCH] remove ReceiveEvent --- cluster.go | 152 ----------------------------------------- pilosa.go | 24 ------- server.go | 5 -- utils_internal_test.go | 34 --------- 4 files changed, 215 deletions(-) diff --git a/cluster.go b/cluster.go index 8418bf1c9..51aff4e4d 100644 --- a/cluster.go +++ b/cluster.go @@ -923,37 +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) sendTo(node *topology.Node, m Message) error { if err := c.broadcaster.SendTo(node, m); err != nil { return errors.Wrap(err, "sending") @@ -1659,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() diff --git a/pilosa.go b/pilosa.go index ee633bd52..c98f886e0 100644 --- a/pilosa.go +++ b/pilosa.go @@ -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() } diff --git a/server.go b/server.go index bb62645cb..9588e0ca2 100644 --- a/server.go +++ b/server.go @@ -863,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: diff --git a/utils_internal_test.go b/utils_internal_test.go index d2215b872..823afb2d4 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -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 }