From 6e4ea21ce574404bb4d127ae950fb32b6c7767fa Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 3 Feb 2021 22:34:45 -0600 Subject: [PATCH 1/3] remove gossip listenForJoins --- cluster.go | 160 ----------------------------------------- server.go | 7 -- utils_internal_test.go | 7 -- 3 files changed, 174 deletions(-) diff --git a/cluster.go b/cluster.go index c4a295072..8418bf1c9 100644 --- a/cluster.go +++ b/cluster.go @@ -954,61 +954,6 @@ func (c *cluster) allNodesReady() (ret bool) { 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 +961,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 +1337,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 +1357,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) } diff --git a/server.go b/server.go index 71262c829..5badf260f 100644 --- a/server.go +++ b/server.go @@ -617,13 +617,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 { diff --git a/utils_internal_test.go b/utils_internal_test.go index 0f2d58cff..d2215b872 100644 --- a/utils_internal_test.go +++ b/utils_internal_test.go @@ -338,13 +338,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 } From 652014539c6fa10333e55f07f2c5becdaab941fc Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 3 Feb 2021 22:36:38 -0600 Subject: [PATCH 2/3] remove temporary Gossiper interface --- server.go | 14 -------------- server/cluster_test.go | 18 +++++++++--------- server/server.go | 7 ------- test/pilosa.go | 7 ------- translator_test.go | 2 +- 5 files changed, 10 insertions(+), 38 deletions(-) diff --git a/server.go b/server.go index 5badf260f..bb62645cb 100644 --- a/server.go +++ b/server.go @@ -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") diff --git a/server/cluster_test.go b/server/cluster_test.go index 8700d6bc6..23668c64f 100644 --- a/server/cluster_test.go +++ b/server/cluster_test.go @@ -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 diff --git a/server/server.go b/server/server.go index 5282e3d6d..00ab4d547 100644 --- a/server/server.go +++ b/server/server.go @@ -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 { diff --git a/test/pilosa.go b/test/pilosa.go index 92e82e077..13993e199 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -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 diff --git a/translator_test.go b/translator_test.go index d8fd79d82..ddaff794b 100644 --- a/translator_test.go +++ b/translator_test.go @@ -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 { From afc53e1163c969b5aa304e7cf776c69bb64314c7 Mon Sep 17 00:00:00 2001 From: Travis Date: Wed, 3 Feb 2021 23:31:38 -0600 Subject: [PATCH 3/3] 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 }