Merge pull request #93 from rstudio/fix-leader-ping

Fix unhandled ping requests from the same PgxLeader node

Author: glin

pkg/rselection/impls/pgx/leader.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+
 	"github.com/rstudio/platform-lib/pkg/rselection"
 	"github.com/rstudio/platform-lib/pkg/rselection/electiontypes"
 	"github.com/rstudio/platform-lib/pkg/rsnotify/broadcaster"
@@ -134,10 +135,10 @@ func (p *PgxLeader) lead(pingTick, sweepTick <-chan time.Time, stop chan bool) {
 	l := p.awb.Subscribe(electiontypes.ClusterMessageTypePingResponse)
 	defer p.awb.Unsubscribe(l)
 
-	// Listen for other ping requests. If an expected ping request is received,
+	// Listen for other ping requests. If an unexpected ping request is received,
 	// we must demote the leader since there's another leader.
-	unexpectedPings := p.awb.Subscribe(electiontypes.ClusterMessageTypePing)
-	defer p.awb.Unsubscribe(unexpectedPings)
+	leaderPings := p.awb.Subscribe(electiontypes.ClusterMessageTypePing)
+	defer p.awb.Unsubscribe(leaderPings)
 
 	// Listen for requests to enumerate nodes in the cluster.
 	nodesCh := p.awb.Subscribe(electiontypes.ClusterMessageTypeNodes)
@@ -171,9 +172,9 @@ func (p *PgxLeader) lead(pingTick, sweepTick <-chan time.Time, stop chan bool) {
 			if cn, ok := n.(*electiontypes.ClusterPingResponse); ok {
 				go p.handlePingResponse(cn)
 			}
-		case n := <-unexpectedPings:
+		case n := <-leaderPings:
 			if cn, ok := n.(*electiontypes.ClusterPingRequest); ok {
-				go p.handleUnexpectedPing(cn)
+				go p.handleLeaderPing(cn)
 			}
 		case n := <-nodesCh:
 			// This supports receiving a request to enumerate cluster nodes. The leader
@@ -215,12 +216,12 @@ func (p *PgxLeader) info() string {
 // this to ensure that the cluster is healthy before running scheduled tasks.
 // This helps to prevent split-brain issues in the cluster.
 //
-// 1. The task handler is ready to run a scheduled task.
-// 2. The task handler sends a `chan bool` to its verify channel.
-// 3. We receive the `chan bool` here and:
-//    a. Verify that the cluster is healthy.
-//    b. Respond with `true` over the channel if the cluster is healthy.
-// 4. The task handler runs the scheduled task when the cluster is healthy.
+//  1. The task handler is ready to run a scheduled task.
+//  2. The task handler sends a `chan bool` to its verify channel.
+//  3. We receive the `chan bool` here and:
+//     a. Verify that the cluster is healthy.
+//     b. Respond with `true` over the channel if the cluster is healthy.
+//  4. The task handler runs the scheduled task when the cluster is healthy.
 func (p *PgxLeader) verify(vCh chan bool) {
 	var err error
 
@@ -329,11 +330,14 @@ func (p *PgxLeader) handleNodesRequest(cn *electiontypes.ClusterNodesRequest) {
 	}
 }
 
-func (p *PgxLeader) handleUnexpectedPing(cn *electiontypes.ClusterPingRequest) {
+func (p *PgxLeader) handleLeaderPing(cn *electiontypes.ClusterPingRequest) {
 	// If we received a ping from another leader, then stop leading
 	if cn.SrcAddr != p.address {
 		p.debugLogger.Debugf("Leader received ping from another leader. Stopping and moving back to the follower loop.")
 		p.stop <- true
+	} else {
+		resp := electiontypes.NewClusterPingResponse(p.address, cn.SrcAddr, p.awb.IP())
+		p.handlePingResponse(resp)
 	}
 }
 

pkg/rselection/impls/pgx/leader_test.go

@@ -10,12 +10,13 @@ import (
 
 	"github.com/fortytw2/leaktest"
 	"github.com/jackc/pgx/v4/pgxpool"
+	"gopkg.in/check.v1"
+
 	"github.com/rstudio/platform-lib/pkg/rselection"
 	"github.com/rstudio/platform-lib/pkg/rselection/electiontypes"
 	"github.com/rstudio/platform-lib/pkg/rsnotify/broadcaster"
 	"github.com/rstudio/platform-lib/pkg/rsnotify/listener"
 	"github.com/rstudio/platform-lib/pkg/rsnotify/listeners/postgrespgx"
-	"gopkg.in/check.v1"
 )
 
 type fakeTaskHandler struct {
@@ -363,6 +364,84 @@ func (s *LeaderSuite) TestPingNodes(c *check.C) {
 	c.Check(leader.unsuccessfulPing(), check.Equals, true)
 }
 
+func (s *LeaderSuite) TestLeaderPingSelf(c *check.C) {
+	defer leaktest.Check(c)
+
+	channel := c.TestName()
+	// Use a real notifier to send the Ping Request notification.
+	realNotifier := &PgxPgNotifier{pool: s.pool}
+	// Use a fake notifier to record the Ping Response notification.
+	fakeNotifier := &dummyNotifier{}
+	matcher := listener.NewMatcher("MessageType")
+	matcher.Register(electiontypes.ClusterMessageTypePingResponse, &electiontypes.ClusterPingResponse{})
+	matcher.Register(electiontypes.ClusterMessageTypePing, &electiontypes.ClusterPingRequest{})
+	matcher.Register(electiontypes.ClusterMessageTypeNodes, &electiontypes.ClusterNodesNotification{})
+	plf := postgrespgx.NewPgxListener(postgrespgx.PgxListenerArgs{
+		Name:       channel + "_leader",
+		Pool:       s.pool,
+		Matcher:    matcher,
+		IpReporter: &listener.TestIPReporter{Ip: "192.168.5.11"},
+	})
+	defer plf.Stop()
+	awbStop := make(chan bool)
+	awb, err := broadcaster.NewNotificationBroadcaster(plf, awbStop)
+	c.Assert(err, check.IsNil)
+	defer func() {
+		awbStop <- true
+	}()
+	stop := make(chan bool)
+
+	pingCh := make(chan bool)
+	defer close(pingCh)
+
+	leader := &PgxLeader{
+		awb:                awb,
+		notify:             fakeNotifier,
+		chLeader:           "leader",
+		chFollower:         "follower",
+		address:            "leader",
+		stop:               stop,
+		nodes:              map[string]*electiontypes.ClusterNode{},
+		pingResponseChTEST: pingCh,
+		taskHandler:        &fakeTaskHandler{},
+		debugLogger:        &fakeLogger{},
+		traceLogger:        &fakeLogger{},
+	}
+
+	// Notified when leader exits
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		leader.lead(nil, nil, stop)
+	}()
+
+	// Receive notification from self
+	msgBytes, err := json.Marshal(&electiontypes.ClusterPingRequest{
+		ClusterNotification: electiontypes.ClusterNotification{
+			GuidVal:     "65db0d7d-8db1-4fa8-bc2a-58fad248507f",
+			MessageType: electiontypes.ClusterMessageTypePing,
+			SrcAddr:     "leader",
+		},
+	})
+	c.Assert(err, check.IsNil)
+
+	now := time.Now()
+	wait(pingCh, func() {
+		err = realNotifier.Notify(channel+"_leader", msgBytes)
+		c.Assert(err, check.IsNil)
+	})
+	c.Assert(len(leader.nodes), check.Equals, 1)
+	node, ok := leader.nodes["leader_192.168.5.11"]
+	c.Assert(ok, check.Equals, true)
+	c.Assert(node.Ping.After(now), check.Equals, true)
+
+	// Stop
+	close(stop)
+
+	// Wait for exit
+	<-done
+}
+
 func (s *LeaderSuite) TestLeaderDemotion(c *check.C) {
 	defer leaktest.Check(c)