WIP ShowReplicationStatus with context support

go/mysql/endtoend/client_test.go

@@ -287,7 +287,28 @@ func TestReplicationStatus(t *testing.T) {
 
 	status, err := conn.ShowReplicationStatus()
 	assert.Equal(t, mysql.ErrNotReplica, err, "Got unexpected result for ShowReplicationStatus: %v %v", status, err)
+}
+
+func TestReplicationStatusWithMysqlHang(t *testing.T) {
+	params := connParams
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+
+	conn, err := mysql.Connect(ctx, &params)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	err = cluster.SimulateMySQLHang()
+	require.NoError(t, err)
+
+	defer cluster.StopSimulateMySQLHang()
 
+	status, err := conn.ShowReplicationStatusWithContext(ctx)
+	assert.Equal(t, ctx.Err().Error(), "context deadline exceeded")
+	assert.Equal(t, ctx.Err(), err, "Got unexpected result for ShowReplicationStatus: %v %v", status, err)
+	assert.True(t, conn.IsClosed())
 }
 
 func TestSessionTrackGTIDs(t *testing.T) {

go/mysql/endtoend/main_test.go

@@ -40,6 +40,7 @@ import (
 
 var (
 	connParams mysql.ConnParams
+	cluster    vttest.LocalCluster
 )
 
 // assertSQLError makes sure we get the right error.
@@ -200,8 +201,17 @@ ssl-key=%v/server-key.pem
 			OnlyMySQL:  true,
 			ExtraMyCnf: []string{extraMyCnf, maxPacketMyCnf},
 		}
-		cluster := vttest.LocalCluster{
+
+		env, err := vttest.NewLocalTestEnv(0)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "%v", err)
+			return 1
+		}
+		env.EnableToxiproxy = true
+
+		cluster = vttest.LocalCluster{
 			Config: cfg,
+			Env:    env,
 		}
 		if err := cluster.Setup(); err != nil {
 			fmt.Fprintf(os.Stderr, "could not launch mysql: %v\n", err)

go/mysql/flavor.go

@@ -441,6 +441,34 @@ func (c *Conn) ShowReplicationStatus() (replication.ReplicationStatus, error) {
 	return c.flavor.status(c)
 }
 
+func (c *Conn) ShowReplicationStatusWithContext(ctx context.Context) (replication.ReplicationStatus, error) {
+	result := make(chan replication.ReplicationStatus, 1)
+	errors := make(chan error, 1)
+
+	go func() {
+		res, err := c.ShowReplicationStatus()
+		if err != nil {
+			errors <- err
+		} else {
+			result <- res
+		}
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			c.Close()
+			return replication.ReplicationStatus{}, ctx.Err()
+
+		case err := <-errors:
+			return replication.ReplicationStatus{}, err
+
+		case res := <-result:
+			return res, nil
+		}
+	}
+}
+
 // ShowPrimaryStatus executes the right SHOW MASTER STATUS command,
 // and returns a parsed executed Position, as well as file based Position.
 func (c *Conn) ShowPrimaryStatus() (replication.PrimaryStatus, error) {

go/vt/mysqlctl/replication.go

@@ -385,13 +385,20 @@ func (mysqld *Mysqld) WaitSourcePos(ctx context.Context, targetPos replication.P
 
 // ReplicationStatus returns the server replication status
 func (mysqld *Mysqld) ReplicationStatus() (replication.ReplicationStatus, error) {
-	conn, err := getPoolReconnect(context.TODO(), mysqld.dbaPool)
+	ctx, cancel := context.WithTimeout(context.Background(), 1000*time.Millisecond)
+	defer cancel()
+
+	return mysqld.ReplicationStatusWithContext(ctx)
+}
+
+func (mysqld *Mysqld) ReplicationStatusWithContext(ctx context.Context) (replication.ReplicationStatus, error) {
+	conn, err := getPoolReconnect(ctx, mysqld.dbaPool)
 	if err != nil {
 		return replication.ReplicationStatus{}, err
 	}
-	defer conn.Recycle()
 
-	return conn.Conn.ShowReplicationStatus()
+	defer conn.Recycle()
+	return conn.Conn.ShowReplicationStatusWithContext(ctx)
 }
 
 // PrimaryStatus returns the primary replication statuses