Adds RPCs to vttablet that vtorc requires

go/mysql/flavor.go

@@ -75,6 +75,15 @@ type flavor interface {
 	// primaryGTIDSet returns the current GTIDSet of a server.
 	primaryGTIDSet(c *Conn) (GTIDSet, error)
 
+	// purgedGTIDSet returns the purged GTIDSet of a server.
+	purgedGTIDSet(c *Conn) (GTIDSet, error)
+
+	// gtidMode returns the gtid mode of a server.
+	gtidMode(c *Conn) (string, error)
+
+	// serverUUID returns the UUID of a server.
+	serverUUID(c *Conn) (string, error)
+
 	// startReplicationCommand returns the command to start the replication.
 	startReplicationCommand() string
 
@@ -112,6 +121,10 @@ type flavor interface {
 	// replication on the host.
 	resetReplicationCommands(c *Conn) []string
 
+	// resetReplicationParametersCommands returns the commands to reset
+	// replication parameters on the host.
+	resetReplicationParametersCommands(c *Conn) []string
+
 	// setReplicationPositionCommands returns the commands to set the
 	// replication position at which the replica will resume.
 	setReplicationPositionCommands(pos Position) []string
@@ -266,6 +279,27 @@ func (c *Conn) PrimaryPosition() (Position, error) {
 	}, nil
 }
 
+// GetGTIDPurged returns the tablet's GTIDs which are purged.
+func (c *Conn) GetGTIDPurged() (Position, error) {
+	gtidSet, err := c.flavor.purgedGTIDSet(c)
+	if err != nil {
+		return Position{}, err
+	}
+	return Position{
+		GTIDSet: gtidSet,
+	}, nil
+}
+
+// GetGTIDMode returns the tablet's GTID mode. Only available in MySQL flavour
+func (c *Conn) GetGTIDMode() (string, error) {
+	return c.flavor.gtidMode(c)
+}
+
+// GetServerUUID returns the server's UUID.
+func (c *Conn) GetServerUUID() (string, error) {
+	return c.flavor.serverUUID(c)
+}
+
 // PrimaryFilePosition returns the current primary's file based replication position.
 func (c *Conn) PrimaryFilePosition() (Position, error) {
 	filePosFlavor := filePosFlavor{}
@@ -339,6 +373,12 @@ func (c *Conn) ResetReplicationCommands() []string {
 	return c.flavor.resetReplicationCommands(c)
 }
 
+// ResetReplicationParametersCommands returns the commands to reset
+// replication parameters on the host.
+func (c *Conn) ResetReplicationParametersCommands() []string {
+	return c.flavor.resetReplicationParametersCommands(c)
+}
+
 // SetReplicationPositionCommands returns the commands to set the
 // replication position at which the replica will resume
 // when it is later reparented with SetReplicationSourceCommand.
@@ -402,7 +442,12 @@ func parseReplicationStatus(fields map[string]string) ReplicationStatus {
 	// The field names in the map are identical to what we receive from the database
 	// Hence the names still contain Master
 	status := ReplicationStatus{
-		SourceHost: fields["Master_Host"],
+		SourceHost:            fields["Master_Host"],
+		SourceUser:            fields["Master_User"],
+		SSLAllowed:            fields["Master_SSL_Allowed"] == "Yes",
+		AutoPosition:          fields["Auto_Position"] == "1",
+		UsingGTID:             fields["Using_Gtid"] != "No" && fields["Using_Gtid"] != "",
+		HasReplicationFilters: (fields["Replicate_Do_DB"] != "") || (fields["Replicate_Ignore_DB"] != "") || (fields["Replicate_Do_Table"] != "") || (fields["Replicate_Ignore_Table"] != "") || (fields["Replicate_Wild_Do_Table"] != "") || (fields["Replicate_Wild_Ignore_Table"] != ""),
 		// These fields are returned from the underlying DB and cannot be renamed
 		IOState:      ReplicationStatusToState(fields["Slave_IO_Running"]),
 		LastIOError:  fields["Last_IO_Error"],
@@ -424,6 +469,8 @@ func parseReplicationStatus(fields map[string]string) ReplicationStatus {
 	}
 	parseUint, _ = strconv.ParseUint(fields["Master_Server_Id"], 10, 0)
 	status.SourceServerID = uint(parseUint)
+	parseUint, _ = strconv.ParseUint(fields["SQL_Delay"], 10, 0)
+	status.SQLDelay = uint(parseUint)
 
 	executedPosStr := fields["Exec_Master_Log_Pos"]
 	file := fields["Relay_Master_Log_File"]
@@ -442,12 +489,24 @@ func parseReplicationStatus(fields map[string]string) ReplicationStatus {
 	if file != "" && readPosStr != "" {
 		fileRelayPos, err := strconv.Atoi(readPosStr)
 		if err == nil {
-			status.FileRelayLogPosition.GTIDSet = filePosGTID{
+			status.RelayLogSourceBinlogEquivalentPosition.GTIDSet = filePosGTID{
 				file: file,
 				pos:  fileRelayPos,
 			}
 		}
 	}
+
+	relayPosStr := fields["Relay_Log_Pos"]
+	file = fields["Relay_Log_File"]
+	if file != "" && relayPosStr != "" {
+		relayFilePos, err := strconv.Atoi(relayPosStr)
+		if err == nil {
+			status.RelayLogFilePosition.GTIDSet = filePosGTID{
+				file: file,
+				pos:  relayFilePos,
+			}
+		}
+	}
 	return status
 }
 

go/mysql/flavor_filepos.go

@@ -17,13 +17,15 @@ limitations under the License.
 package mysql
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"strconv"
 	"strings"
 	"time"
 
-	"context"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+	"vitess.io/vitess/go/vt/vterrors"
 )
 
 type filePosFlavor struct {
@@ -62,6 +64,36 @@ func (flv *filePosFlavor) primaryGTIDSet(c *Conn) (GTIDSet, error) {
 	}, nil
 }
 
+// purgedGTIDSet is part of the Flavor interface.
+func (flv *filePosFlavor) purgedGTIDSet(c *Conn) (GTIDSet, error) {
+	return nil, nil
+}
+
+// gtidMode is part of the Flavor interface.
+func (flv *filePosFlavor) gtidMode(c *Conn) (string, error) {
+	qr, err := c.ExecuteFetch("select @@global.gtid_mode", 1, false)
+	if err != nil {
+		return "", err
+	}
+	if len(qr.Rows) != 1 || len(qr.Rows[0]) != 1 {
+		return "", vterrors.Errorf(vtrpcpb.Code_INTERNAL, "unexpected result format for gtid_mode: %#v", qr)
+	}
+	return qr.Rows[0][0].ToString(), nil
+}
+
+// serverUUID is part of the Flavor interface.
+func (flv *filePosFlavor) serverUUID(c *Conn) (string, error) {
+	// keep @@global as lowercase, as some servers like the Ripple binlog server only honors a lowercase `global` value
+	qr, err := c.ExecuteFetch("SELECT @@global.server_uuid", 1, false)
+	if err != nil {
+		return "", err
+	}
+	if len(qr.Rows) != 1 || len(qr.Rows[0]) != 1 {
+		return "", vterrors.Errorf(vtrpcpb.Code_INTERNAL, "unexpected result format for server_uuid: %#v", qr)
+	}
+	return qr.Rows[0][0].ToString(), nil
+}
+
 func (flv *filePosFlavor) startReplicationCommand() string {
 	return "unsupported"
 }
@@ -183,6 +215,13 @@ func (flv *filePosFlavor) resetReplicationCommands(c *Conn) []string {
 	}
 }
 
+// resetReplicationParametersCommands is part of the Flavor interface.
+func (flv *filePosFlavor) resetReplicationParametersCommands(c *Conn) []string {
+	return []string{
+		"unsupported",
+	}
+}
+
 // setReplicationPositionCommands is part of the Flavor interface.
 func (flv *filePosFlavor) setReplicationPositionCommands(pos Position) []string {
 	return []string{
@@ -219,7 +258,7 @@ func parseFilePosReplicationStatus(resultMap map[string]string) (ReplicationStat
 	status := parseReplicationStatus(resultMap)
 
 	status.Position = status.FilePosition
-	status.RelayLogPosition = status.FileRelayLogPosition
+	status.RelayLogPosition = status.RelayLogSourceBinlogEquivalentPosition
 
 	return status, nil
 }

go/mysql/flavor_filepos_test.go

@@ -40,22 +40,26 @@ func TestFilePosRetrieveExecutedPosition(t *testing.T) {
 		"Relay_Master_Log_File": "master-bin.000002",
 		"Read_Master_Log_Pos":   "1308",
 		"Master_Log_File":       "master-bin.000003",
+		"Relay_Log_Pos":         "1309",
+		"Relay_Log_File":        "relay-bin.000004",
 	}
 
 	want := ReplicationStatus{
-		Position:             Position{GTIDSet: filePosGTID{file: "master-bin.000002", pos: 1307}},
-		RelayLogPosition:     Position{GTIDSet: filePosGTID{file: "master-bin.000003", pos: 1308}},
-		FilePosition:         Position{GTIDSet: filePosGTID{file: "master-bin.000002", pos: 1307}},
-		FileRelayLogPosition: Position{GTIDSet: filePosGTID{file: "master-bin.000003", pos: 1308}},
+		Position:                               Position{GTIDSet: filePosGTID{file: "master-bin.000002", pos: 1307}},
+		RelayLogPosition:                       Position{GTIDSet: filePosGTID{file: "master-bin.000003", pos: 1308}},
+		FilePosition:                           Position{GTIDSet: filePosGTID{file: "master-bin.000002", pos: 1307}},
+		RelayLogSourceBinlogEquivalentPosition: Position{GTIDSet: filePosGTID{file: "master-bin.000003", pos: 1308}},
+		RelayLogFilePosition:                   Position{GTIDSet: filePosGTID{file: "relay-bin.000004", pos: 1309}},
 	}
 	got, err := parseFilePosReplicationStatus(resultMap)
 	require.NoError(t, err)
 	assert.Equalf(t, got.Position.GTIDSet, want.Position.GTIDSet, "got Position: %v; want Position: %v", got.Position.GTIDSet, want.Position.GTIDSet)
 	assert.Equalf(t, got.RelayLogPosition.GTIDSet, want.RelayLogPosition.GTIDSet, "got RelayLogPosition: %v; want RelayLogPosition: %v", got.RelayLogPosition.GTIDSet, want.RelayLogPosition.GTIDSet)
+	assert.Equalf(t, got.RelayLogFilePosition.GTIDSet, want.RelayLogFilePosition.GTIDSet, "got RelayLogFilePosition: %v; want RelayLogFilePosition: %v", got.RelayLogFilePosition.GTIDSet, want.RelayLogFilePosition.GTIDSet)
 	assert.Equalf(t, got.FilePosition.GTIDSet, want.FilePosition.GTIDSet, "got FilePosition: %v; want FilePosition: %v", got.FilePosition.GTIDSet, want.FilePosition.GTIDSet)
-	assert.Equalf(t, got.FileRelayLogPosition.GTIDSet, want.FileRelayLogPosition.GTIDSet, "got FileRelayLogPosition: %v; want FileRelayLogPosition: %v", got.FileRelayLogPosition.GTIDSet, want.FileRelayLogPosition.GTIDSet)
+	assert.Equalf(t, got.RelayLogSourceBinlogEquivalentPosition.GTIDSet, want.RelayLogSourceBinlogEquivalentPosition.GTIDSet, "got RelayLogSourceBinlogEquivalentPosition: %v; want RelayLogSourceBinlogEquivalentPosition: %v", got.RelayLogSourceBinlogEquivalentPosition.GTIDSet, want.RelayLogSourceBinlogEquivalentPosition.GTIDSet)
 	assert.Equalf(t, got.Position.GTIDSet, got.FilePosition.GTIDSet, "FilePosition and Position don't match when they should for the FilePos flavor")
-	assert.Equalf(t, got.RelayLogPosition.GTIDSet, got.FileRelayLogPosition.GTIDSet, "RelayLogPosition and FileRelayLogPosition don't match when they should for the FilePos flavor")
+	assert.Equalf(t, got.RelayLogPosition.GTIDSet, got.RelayLogSourceBinlogEquivalentPosition.GTIDSet, "RelayLogPosition and RelayLogSourceBinlogEquivalentPosition don't match when they should for the FilePos flavor")
 }
 
 func TestFilePosShouldGetPosition(t *testing.T) {

go/mysql/flavor_mariadb.go

@@ -53,6 +53,21 @@ func (mariadbFlavor) primaryGTIDSet(c *Conn) (GTIDSet, error) {
 	return parseMariadbGTIDSet(qr.Rows[0][0].ToString())
 }
 
+// purgedGTIDSet is part of the Flavor interface.
+func (mariadbFlavor) purgedGTIDSet(c *Conn) (GTIDSet, error) {
+	return nil, nil
+}
+
+// serverUUID is part of the Flavor interface.
+func (mariadbFlavor) serverUUID(c *Conn) (string, error) {
+	return "", nil
+}
+
+// gtidMode is part of the Flavor interface.
+func (mariadbFlavor) gtidMode(c *Conn) (string, error) {
+	return "", nil
+}
+
 func (mariadbFlavor) startReplicationUntilAfter(pos Position) string {
 	return fmt.Sprintf("START SLAVE UNTIL master_gtid_pos = \"%s\"", pos)
 }
@@ -130,6 +145,14 @@ func (mariadbFlavor) resetReplicationCommands(c *Conn) []string {
 	return resetCommands
 }
 
+// resetReplicationParametersCommands is part of the Flavor interface.
+func (mariadbFlavor) resetReplicationParametersCommands(c *Conn) []string {
+	resetCommands := []string{
+		"RESET SLAVE ALL", // "ALL" makes it forget source host:port.
+	}
+	return resetCommands
+}
+
 // setReplicationPositionCommands is part of the Flavor interface.
 func (mariadbFlavor) setReplicationPositionCommands(pos Position) []string {
 	return []string{

go/mysql/flavor_mariadb_test.go

@@ -99,16 +99,20 @@ func TestMariadbRetrieveFileBasedPositions(t *testing.T) {
 		"Read_Master_Log_Pos":   "1308",
 		"Master_Log_File":       "master-bin.000003",
 		"Gtid_Slave_Pos":        "0-101-2320",
+		"Relay_Log_Pos":         "1309",
+		"Relay_Log_File":        "relay-bin.000004",
 	}
 
 	want := ReplicationStatus{
-		FilePosition:         Position{GTIDSet: filePosGTID{file: "master-bin.000002", pos: 1307}},
-		FileRelayLogPosition: Position{GTIDSet: filePosGTID{file: "master-bin.000003", pos: 1308}},
+		FilePosition:                           Position{GTIDSet: filePosGTID{file: "master-bin.000002", pos: 1307}},
+		RelayLogSourceBinlogEquivalentPosition: Position{GTIDSet: filePosGTID{file: "master-bin.000003", pos: 1308}},
+		RelayLogFilePosition:                   Position{GTIDSet: filePosGTID{file: "relay-bin.000004", pos: 1309}},
 	}
 	got, err := parseMariadbReplicationStatus(resultMap)
 	require.NoError(t, err)
+	assert.Equalf(t, got.RelayLogFilePosition.GTIDSet, want.RelayLogFilePosition.GTIDSet, "got RelayLogFilePosition: %v; want RelayLogFilePosition: %v", got.RelayLogFilePosition.GTIDSet, want.RelayLogFilePosition.GTIDSet)
 	assert.Equal(t, got.FilePosition.GTIDSet, want.FilePosition.GTIDSet, fmt.Sprintf("got FilePosition: %v; want FilePosition: %v", got.FilePosition.GTIDSet, want.FilePosition.GTIDSet))
-	assert.Equal(t, got.FileRelayLogPosition.GTIDSet, want.FileRelayLogPosition.GTIDSet, fmt.Sprintf("got FileRelayLogPosition: %v; want FileRelayLogPosition: %v", got.FileRelayLogPosition.GTIDSet, want.FileRelayLogPosition.GTIDSet))
+	assert.Equal(t, got.RelayLogSourceBinlogEquivalentPosition.GTIDSet, want.RelayLogSourceBinlogEquivalentPosition.GTIDSet, fmt.Sprintf("got RelayLogSourceBinlogEquivalentPosition: %v; want RelayLogSourceBinlogEquivalentPosition: %v", got.RelayLogSourceBinlogEquivalentPosition.GTIDSet, want.RelayLogSourceBinlogEquivalentPosition.GTIDSet))
 }
 
 func TestMariadbShouldGetNilRelayLogPosition(t *testing.T) {

go/mysql/flavor_mysql.go

@@ -56,6 +56,44 @@ func (mysqlFlavor) primaryGTIDSet(c *Conn) (GTIDSet, error) {
 	return parseMysql56GTIDSet(qr.Rows[0][0].ToString())
 }
 
+// purgedGTIDSet is part of the Flavor interface.
+func (mysqlFlavor) purgedGTIDSet(c *Conn) (GTIDSet, error) {
+	// keep @@global as lowercase, as some servers like the Ripple binlog server only honors a lowercase `global` value
+	qr, err := c.ExecuteFetch("SELECT @@global.gtid_purged", 1, false)
+	if err != nil {
+		return nil, err
+	}
+	if len(qr.Rows) != 1 || len(qr.Rows[0]) != 1 {
+		return nil, vterrors.Errorf(vtrpc.Code_INTERNAL, "unexpected result format for gtid_purged: %#v", qr)
+	}
+	return parseMysql56GTIDSet(qr.Rows[0][0].ToString())
+}
+
+// serverUUID is part of the Flavor interface.
+func (mysqlFlavor) serverUUID(c *Conn) (string, error) {
+	// keep @@global as lowercase, as some servers like the Ripple binlog server only honors a lowercase `global` value
+	qr, err := c.ExecuteFetch("SELECT @@global.server_uuid", 1, false)
+	if err != nil {
+		return "", err
+	}
+	if len(qr.Rows) != 1 || len(qr.Rows[0]) != 1 {
+		return "", vterrors.Errorf(vtrpc.Code_INTERNAL, "unexpected result format for server_uuid: %#v", qr)
+	}
+	return qr.Rows[0][0].ToString(), nil
+}
+
+// gtidMode is part of the Flavor interface.
+func (mysqlFlavor) gtidMode(c *Conn) (string, error) {
+	qr, err := c.ExecuteFetch("select @@global.gtid_mode", 1, false)
+	if err != nil {
+		return "", err
+	}
+	if len(qr.Rows) != 1 || len(qr.Rows[0]) != 1 {
+		return "", vterrors.Errorf(vtrpc.Code_INTERNAL, "unexpected result format for gtid_mode: %#v", qr)
+	}
+	return qr.Rows[0][0].ToString(), nil
+}
+
 func (mysqlFlavor) startReplicationCommand() string {
 	return "START SLAVE"
 }
@@ -117,6 +155,14 @@ func (mysqlFlavor) resetReplicationCommands(c *Conn) []string {
 	return resetCommands
 }
 
+// resetReplicationParametersCommands is part of the Flavor interface.
+func (mysqlFlavor) resetReplicationParametersCommands(c *Conn) []string {
+	resetCommands := []string{
+		"RESET SLAVE ALL", // "ALL" makes it forget source host:port.
+	}
+	return resetCommands
+}
+
 // setReplicationPositionCommands is part of the Flavor interface.
 func (mysqlFlavor) setReplicationPositionCommands(pos Position) []string {
 	return []string{

go/mysql/flavor_mysql_test.go

@@ -96,16 +96,20 @@ func TestMysqlRetrieveFileBasedPositions(t *testing.T) {
 		"Relay_Master_Log_File": "master-bin.000002",
 		"Read_Master_Log_Pos":   "1308",
 		"Master_Log_File":       "master-bin.000003",
+		"Relay_Log_Pos":         "1309",
+		"Relay_Log_File":        "relay-bin.000004",
 	}
 
 	want := ReplicationStatus{
-		FilePosition:         Position{GTIDSet: filePosGTID{file: "master-bin.000002", pos: 1307}},
-		FileRelayLogPosition: Position{GTIDSet: filePosGTID{file: "master-bin.000003", pos: 1308}},
+		FilePosition:                           Position{GTIDSet: filePosGTID{file: "master-bin.000002", pos: 1307}},
+		RelayLogSourceBinlogEquivalentPosition: Position{GTIDSet: filePosGTID{file: "master-bin.000003", pos: 1308}},
+		RelayLogFilePosition:                   Position{GTIDSet: filePosGTID{file: "relay-bin.000004", pos: 1309}},
 	}
 	got, err := parseMysqlReplicationStatus(resultMap)
 	require.NoError(t, err)
 	assert.Equalf(t, got.FilePosition.GTIDSet, want.FilePosition.GTIDSet, "got FilePosition: %v; want FilePosition: %v", got.FilePosition.GTIDSet, want.FilePosition.GTIDSet)
-	assert.Equalf(t, got.FileRelayLogPosition.GTIDSet, want.FileRelayLogPosition.GTIDSet, "got FileRelayLogPosition: %v; want FileRelayLogPosition: %v", got.FileRelayLogPosition.GTIDSet, want.FileRelayLogPosition.GTIDSet)
+	assert.Equalf(t, got.RelayLogFilePosition.GTIDSet, want.RelayLogFilePosition.GTIDSet, "got RelayLogFilePosition: %v; want RelayLogFilePosition: %v", got.RelayLogFilePosition.GTIDSet, want.RelayLogFilePosition.GTIDSet)
+	assert.Equalf(t, got.RelayLogSourceBinlogEquivalentPosition.GTIDSet, want.RelayLogSourceBinlogEquivalentPosition.GTIDSet, "got RelayLogSourceBinlogEquivalentPosition: %v; want RelayLogSourceBinlogEquivalentPosition: %v", got.RelayLogSourceBinlogEquivalentPosition.GTIDSet, want.RelayLogSourceBinlogEquivalentPosition.GTIDSet)
 }
 
 func TestMysqlShouldGetRelayLogPosition(t *testing.T) {

go/mysql/flavor_mysqlgr.go

@@ -93,6 +93,11 @@ func (mysqlGRFlavor) resetReplicationCommands(c *Conn) []string {
 	return []string{}
 }
 
+// resetReplicationParametersCommands is part of the Flavor interface.
+func (mysqlGRFlavor) resetReplicationParametersCommands(c *Conn) []string {
+	return []string{}
+}
+
 // setReplicationPositionCommands is disabled in mysqlGRFlavor
 func (mysqlGRFlavor) setReplicationPositionCommands(pos Position) []string {
 	return []string{}

go/mysql/mariadb_gtid.go

@@ -67,6 +67,10 @@ func parseMariadbGTIDSet(s string) (GTIDSet, error) {
 	gtidStrings := strings.Split(s, ",")
 	gtidSet := make(MariadbGTIDSet, len(gtidStrings))
 	for _, gtidString := range gtidStrings {
+		gtidString = strings.TrimSpace(gtidString)
+		if gtidString == "" {
+			continue
+		}
 		gtid, err := parseMariadbGTID(gtidString)
 		if err != nil {
 			return nil, err