Merge pull request #1 from wistia/anders/mysql-8.4-support

Add MySQL 8.4 Support

.github/workflows/replica-tests.yml

@@ -7,6 +7,7 @@ jobs:
 
     runs-on: ubuntu-20.04
     strategy:
+      fail-fast: false
       matrix:
         version: [mysql-5.7.25,mysql-8.0.16,PerconaServer-8.0.21]
 
@@ -22,3 +23,27 @@ jobs:
       env:
         TEST_MYSQL_VERSION: ${{ matrix.version }}
       run:  script/cibuild-gh-ost-replica-tests
+
+  docker-tests:
+    runs-on: ubuntu-22.04
+    strategy:
+      fail-fast: false
+      matrix:
+        image: ['mysql:8.4.3']
+    env:
+      TEST_MYSQL_IMAGE: ${{ matrix.image }}
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Setup environment
+        run: script/docker-gh-ost-replica-tests up
+
+      - name: Run tests
+        run: script/docker-gh-ost-replica-tests run
+
+      - name: Teardown environment
+        if: always()
+        run: script/docker-gh-ost-replica-tests down
+
+

go.mod

@@ -36,6 +36,7 @@ require (
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/hashicorp/go-version v1.7.0 // indirect
 	github.com/klauspost/compress v1.17.4 // indirect
 	github.com/kr/text v0.2.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect

go.sum

@@ -65,6 +65,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 h1:YBftPWNWd4WwGqtY2yeZL2ef8rHAxPBD8KFhJpmcqms=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0/go.mod h1:YN5jB8ie0yfIUg6VvR9Kz84aCaG7AsGZnLjhHbUqwPg=
+github.com/hashicorp/go-version v1.7.0 h1:5tqGy27NaOTB8yJKUZELlFAS/LTKJkrmONwQKeRZfjY=
+github.com/hashicorp/go-version v1.7.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
 github.com/jmoiron/sqlx v1.3.3/go.mod h1:2BljVx/86SuTyjE+aPYlHCTNvZrnJXghYGpNiXLBMCQ=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=

go/logic/applier.go

@@ -829,7 +829,8 @@ func (this *Applier) RenameTablesRollback() (renameError error) {
 // We need to keep the SQL thread active so as to complete processing received events,
 // and have them written to the binary log, so that we can then read them via streamer.
 func (this *Applier) StopSlaveIOThread() error {
-	query := `stop /* gh-ost */ slave io_thread`
+	replicaTerm := mysql.ReplicaTermFor(this.migrationContext.ApplierMySQLVersion, `slave`)
+	query := fmt.Sprintf("stop /* gh-ost */ %s io_thread", replicaTerm)
 	this.migrationContext.Log.Infof("Stopping replication IO thread")
 	if _, err := sqlutils.ExecNoPrepare(this.db, query); err != nil {
 		return err
@@ -840,7 +841,8 @@ func (this *Applier) StopSlaveIOThread() error {
 
 // StartSlaveIOThread is applicable with --test-on-replica
 func (this *Applier) StartSlaveIOThread() error {
-	query := `start /* gh-ost */ slave io_thread`
+	replicaTerm := mysql.ReplicaTermFor(this.migrationContext.ApplierMySQLVersion, `slave`)
+	query := fmt.Sprintf("start /* gh-ost */ %s io_thread", replicaTerm)
 	this.migrationContext.Log.Infof("Starting replication IO thread")
 	if _, err := sqlutils.ExecNoPrepare(this.db, query); err != nil {
 		return err
@@ -851,7 +853,8 @@ func (this *Applier) StartSlaveIOThread() error {
 
 // StopSlaveSQLThread is applicable with --test-on-replica
 func (this *Applier) StopSlaveSQLThread() error {
-	query := `stop /* gh-ost */ slave sql_thread`
+	replicaTerm := mysql.ReplicaTermFor(this.migrationContext.ApplierMySQLVersion, `slave`)
+	query := fmt.Sprintf("stop /* gh-ost */ %s sql_thread", replicaTerm)
 	this.migrationContext.Log.Infof("Verifying SQL thread is stopped")
 	if _, err := sqlutils.ExecNoPrepare(this.db, query); err != nil {
 		return err
@@ -862,7 +865,8 @@ func (this *Applier) StopSlaveSQLThread() error {
 
 // StartSlaveSQLThread is applicable with --test-on-replica
 func (this *Applier) StartSlaveSQLThread() error {
-	query := `start /* gh-ost */ slave sql_thread`
+	replicaTerm := mysql.ReplicaTermFor(this.migrationContext.ApplierMySQLVersion, `slave`)
+	query := fmt.Sprintf("start /* gh-ost */ %s sql_thread", replicaTerm)
 	this.migrationContext.Log.Infof("Verifying SQL thread is running")
 	if _, err := sqlutils.ExecNoPrepare(this.db, query); err != nil {
 		return err
@@ -880,7 +884,7 @@ func (this *Applier) StopReplication() error {
 		return err
 	}
 
-	readBinlogCoordinates, executeBinlogCoordinates, err := mysql.GetReplicationBinlogCoordinates(this.db)
+	readBinlogCoordinates, executeBinlogCoordinates, err := mysql.GetReplicationBinlogCoordinates(this.migrationContext.ApplierMySQLVersion, this.db)
 	if err != nil {
 		return err
 	}

go/logic/inspect.go

@@ -30,6 +30,7 @@ const startReplicationMaxWait = 2 * time.Second
 type Inspector struct {
 	connectionConfig    *mysql.ConnectionConfig
 	db                  *gosql.DB
+	dbVersion           string
 	informationSchemaDb *gosql.DB
 	migrationContext    *base.MigrationContext
 	name                string
@@ -57,6 +58,8 @@ func (this *Inspector) InitDBConnections() (err error) {
 	if err := this.validateConnection(); err != nil {
 		return err
 	}
+	this.dbVersion = this.migrationContext.InspectorMySQLVersion
+
 	if !this.migrationContext.AliyunRDS && !this.migrationContext.GoogleCloudPlatform && !this.migrationContext.AzureMySQL {
 		if impliedKey, err := mysql.GetInstanceKey(this.db); err != nil {
 			return err
@@ -288,15 +291,16 @@ func (this *Inspector) validateGrants() error {
 func (this *Inspector) restartReplication() error {
 	this.migrationContext.Log.Infof("Restarting replication on %s to make sure binlog settings apply to replication thread", this.connectionConfig.Key.String())
 
-	masterKey, _ := mysql.GetMasterKeyFromSlaveStatus(this.connectionConfig)
+	masterKey, _ := mysql.GetMasterKeyFromSlaveStatus(this.dbVersion, this.connectionConfig)
 	if masterKey == nil {
 		// This is not a replica
 		return nil
 	}
 
 	var stopError, startError error
-	_, stopError = sqlutils.ExecNoPrepare(this.db, `stop slave`)
-	_, startError = sqlutils.ExecNoPrepare(this.db, `start slave`)
+	replicaTerm := mysql.ReplicaTermFor(this.dbVersion, `slave`)
+	_, stopError = sqlutils.ExecNoPrepare(this.db, fmt.Sprintf("stop %s", replicaTerm))
+	_, startError = sqlutils.ExecNoPrepare(this.db, fmt.Sprintf("start %s", replicaTerm))
 	if stopError != nil {
 		return stopError
 	}
@@ -329,9 +333,11 @@ func (this *Inspector) restartReplication() error {
 // returns true if both are 'Yes', false otherwise
 func (this *Inspector) validateReplicationRestarted() (bool, error) {
 	errNotRunning := fmt.Errorf("Replication not running on %s", this.connectionConfig.Key.String())
-	query := `show /* gh-ost */ slave status`
+	query := fmt.Sprintf("show /* gh-ost */ %s", mysql.ReplicaTermFor(this.dbVersion, "slave status"))
 	err := sqlutils.QueryRowsMap(this.db, query, func(rowMap sqlutils.RowMap) error {
-		if rowMap.GetString("Slave_IO_Running") != "Yes" || rowMap.GetString("Slave_SQL_Running") != "Yes" {
+		ioRunningTerm := mysql.ReplicaTermFor(this.dbVersion, "Slave_IO_Running")
+		sqlRunningTerm := mysql.ReplicaTermFor(this.dbVersion, "Slave_SQL_Running")
+		if rowMap.GetString(ioRunningTerm) != "Yes" || rowMap.GetString(sqlRunningTerm) != "Yes" {
 			return errNotRunning
 		}
 		return nil
@@ -389,7 +395,7 @@ func (this *Inspector) validateBinlogs() error {
 		if !this.migrationContext.SwitchToRowBinlogFormat {
 			return fmt.Errorf("You must be using ROW binlog format. I can switch it for you, provided --switch-to-rbr and that %s doesn't have replicas", this.connectionConfig.Key.String())
 		}
-		query := `show /* gh-ost */ slave hosts`
+		query := fmt.Sprintf("show /* gh-ost */ %s", mysql.ReplicaTermFor(this.dbVersion, `slave hosts`))
 		countReplicas := 0
 		err := sqlutils.QueryRowsMap(this.db, query, func(rowMap sqlutils.RowMap) error {
 			countReplicas++
@@ -864,11 +870,12 @@ func (this *Inspector) readChangelogState(hint string) (string, error) {
 func (this *Inspector) getMasterConnectionConfig() (applierConfig *mysql.ConnectionConfig, err error) {
 	this.migrationContext.Log.Infof("Recursively searching for replication master")
 	visitedKeys := mysql.NewInstanceKeyMap()
-	return mysql.GetMasterConnectionConfigSafe(this.connectionConfig, visitedKeys, this.migrationContext.AllowedMasterMaster)
+	return mysql.GetMasterConnectionConfigSafe(this.dbVersion, this.connectionConfig, visitedKeys, this.migrationContext.AllowedMasterMaster)
 }
 
 func (this *Inspector) getReplicationLag() (replicationLag time.Duration, err error) {
 	replicationLag, err = mysql.GetReplicationLagFromSlaveStatus(
+		this.dbVersion,
 		this.informationSchemaDb,
 	)
 	return replicationLag, err

go/logic/streamer.go

@@ -36,6 +36,7 @@ const (
 type EventsStreamer struct {
 	connectionConfig         *mysql.ConnectionConfig
 	db                       *gosql.DB
+	dbVersion                string
 	migrationContext         *base.MigrationContext
 	initialBinlogCoordinates *mysql.BinlogCoordinates
 	listeners                [](*BinlogEventListener)
@@ -107,9 +108,11 @@ func (this *EventsStreamer) InitDBConnections() (err error) {
 	if this.db, _, err = mysql.GetDB(this.migrationContext.Uuid, EventsStreamerUri); err != nil {
 		return err
 	}
-	if _, err := base.ValidateConnection(this.db, this.connectionConfig, this.migrationContext, this.name); err != nil {
+	version, err := base.ValidateConnection(this.db, this.connectionConfig, this.migrationContext, this.name)
+	if err != nil {
 		return err
 	}
+	this.dbVersion = version
 	if err := this.readCurrentBinlogCoordinates(); err != nil {
 		return err
 	}
@@ -140,7 +143,8 @@ func (this *EventsStreamer) GetReconnectBinlogCoordinates() *mysql.BinlogCoordin
 
 // readCurrentBinlogCoordinates reads master status from hooked server
 func (this *EventsStreamer) readCurrentBinlogCoordinates() error {
-	query := `show /* gh-ost readCurrentBinlogCoordinates */ master status`
+	binaryLogStatusTerm := mysql.ReplicaTermFor(this.dbVersion, "master status")
+	query := fmt.Sprintf("show /* gh-ost readCurrentBinlogCoordinates */ %s", binaryLogStatusTerm)
 	foundMasterStatus := false
 	err := sqlutils.QueryRowsMap(this.db, query, func(m sqlutils.RowMap) error {
 		this.initialBinlogCoordinates = &mysql.BinlogCoordinates{
@@ -155,7 +159,7 @@ func (this *EventsStreamer) readCurrentBinlogCoordinates() error {
 		return err
 	}
 	if !foundMasterStatus {
-		return fmt.Errorf("Got no results from SHOW MASTER STATUS. Bailing out")
+		return fmt.Errorf("Got no results from SHOW %s. Bailing out", strings.ToUpper(binaryLogStatusTerm))
 	}
 	this.migrationContext.Log.Debugf("Streamer binlog coordinates: %+v", *this.initialBinlogCoordinates)
 	return nil

go/logic/throttler.go

@@ -150,7 +150,7 @@ func (this *Throttler) collectReplicationLag(firstThrottlingCollected chan<- boo
 			// when running on replica, the heartbeat injection is also done on the replica.
 			// This means we will always get a good heartbeat value.
 			// When running on replica, we should instead check the `SHOW SLAVE STATUS` output.
-			if lag, err := mysql.GetReplicationLagFromSlaveStatus(this.inspector.informationSchemaDb); err != nil {
+			if lag, err := mysql.GetReplicationLagFromSlaveStatus(this.inspector.dbVersion, this.inspector.informationSchemaDb); err != nil {
 				return this.migrationContext.Log.Errore(err)
 			} else {
 				atomic.StoreInt64(&this.migrationContext.CurrentLag, int64(lag))

go/mysql/replica_terminology_map.go

@@ -0,0 +1,39 @@
+package mysql
+
+import (
+	version "github.com/hashicorp/go-version"
+)
+
+const (
+	MysqlVersionCutoff = "8.4"
+)
+
+var MysqlReplicaTermMap = map[string]string{
+	"Seconds_Behind_Master": "Seconds_Behind_Source",
+	"Master_Log_File":       "Source_Log_File",
+	"Master_Host":           "Source_Host",
+	"Master_Port":           "Source_Port",
+	"Exec_Master_Log_Pos":   "Exec_Source_Log_Pos",
+	"Read_Master_Log_Pos":   "Read_Source_Log_Pos",
+	"Relay_Master_Log_File": "Relay_Source_Log_File",
+	"Slave_IO_Running":      "Replica_IO_Running",
+	"Slave_SQL_Running":     "Replica_SQL_Running",
+	"master status":         "binary log status",
+	"slave hosts":           "replicas",
+	"slave status":          "replica status",
+	"slave":                 "replica",
+}
+
+func ReplicaTermFor(mysqlVersion string, term string) string {
+	vs, err := version.NewVersion(mysqlVersion)
+	if err != nil {
+		// default to returning the same term if we cannot determine the version
+		return term
+	}
+
+	mysqlVersionCutoff, _ := version.NewVersion(MysqlVersionCutoff)
+	if vs.GreaterThanOrEqual(mysqlVersionCutoff) {
+		return MysqlReplicaTermMap[term]
+	}
+	return term
+}

go/mysql/utils.go

@@ -61,13 +61,16 @@ func GetDB(migrationUuid string, mysql_uri string) (db *gosql.DB, exists bool, e
 }
 
 // GetReplicationLagFromSlaveStatus returns replication lag for a given db; via SHOW SLAVE STATUS
-func GetReplicationLagFromSlaveStatus(informationSchemaDb *gosql.DB) (replicationLag time.Duration, err error) {
-	err = sqlutils.QueryRowsMap(informationSchemaDb, `show slave status`, func(m sqlutils.RowMap) error {
-		slaveIORunning := m.GetString("Slave_IO_Running")
-		slaveSQLRunning := m.GetString("Slave_SQL_Running")
-		secondsBehindMaster := m.GetNullInt64("Seconds_Behind_Master")
+func GetReplicationLagFromSlaveStatus(dbVersion string, informationSchemaDb *gosql.DB) (replicationLag time.Duration, err error) {
+	showReplicaStatusQuery := fmt.Sprintf("show %s", ReplicaTermFor(dbVersion, `slave status`))
+	err = sqlutils.QueryRowsMap(informationSchemaDb, showReplicaStatusQuery, func(m sqlutils.RowMap) error {
+		ioRunningTerm := ReplicaTermFor(dbVersion, "Slave_IO_Running")
+		sqlRunningTerm := ReplicaTermFor(dbVersion, "Slave_SQL_Running")
+		slaveIORunning := m.GetString(ioRunningTerm)
+		slaveSQLRunning := m.GetString(sqlRunningTerm)
+		secondsBehindMaster := m.GetNullInt64(ReplicaTermFor(dbVersion, "Seconds_Behind_Master"))
 		if !secondsBehindMaster.Valid {
-			return fmt.Errorf("replication not running; Slave_IO_Running=%+v, Slave_SQL_Running=%+v", slaveIORunning, slaveSQLRunning)
+			return fmt.Errorf("replication not running; %s=%+v, %s=%+v", ioRunningTerm, slaveIORunning, sqlRunningTerm, slaveSQLRunning)
 		}
 		replicationLag = time.Duration(secondsBehindMaster.Int64) * time.Second
 		return nil
@@ -76,7 +79,7 @@ func GetReplicationLagFromSlaveStatus(informationSchemaDb *gosql.DB) (replicatio
 	return replicationLag, err
 }
 
-func GetMasterKeyFromSlaveStatus(connectionConfig *ConnectionConfig) (masterKey *InstanceKey, err error) {
+func GetMasterKeyFromSlaveStatus(dbVersion string, connectionConfig *ConnectionConfig) (masterKey *InstanceKey, err error) {
 	currentUri := connectionConfig.GetDBUri("information_schema")
 	// This function is only called once, okay to not have a cached connection pool
 	db, err := gosql.Open("mysql", currentUri)
@@ -85,40 +88,45 @@ func GetMasterKeyFromSlaveStatus(connectionConfig *ConnectionConfig) (masterKey
 	}
 	defer db.Close()
 
-	err = sqlutils.QueryRowsMap(db, `show slave status`, func(rowMap sqlutils.RowMap) error {
+	showReplicaStatusQuery := fmt.Sprintf("show %s", ReplicaTermFor(dbVersion, `slave status`))
+	err = sqlutils.QueryRowsMap(db, showReplicaStatusQuery, func(rowMap sqlutils.RowMap) error {
 		// We wish to recognize the case where the topology's master actually has replication configuration.
 		// This can happen when a DBA issues a `RESET SLAVE` instead of `RESET SLAVE ALL`.
 
 		// An empty log file indicates this is a master:
-		if rowMap.GetString("Master_Log_File") == "" {
+		if rowMap.GetString(ReplicaTermFor(dbVersion, "Master_Log_File")) == "" {
 			return nil
 		}
 
-		slaveIORunning := rowMap.GetString("Slave_IO_Running")
-		slaveSQLRunning := rowMap.GetString("Slave_SQL_Running")
+		ioRunningTerm := ReplicaTermFor(dbVersion, "Slave_IO_Running")
+		sqlRunningTerm := ReplicaTermFor(dbVersion, "Slave_SQL_Running")
+		slaveIORunning := rowMap.GetString(ioRunningTerm)
+		slaveSQLRunning := rowMap.GetString(sqlRunningTerm)
 
 		if slaveIORunning != "Yes" || slaveSQLRunning != "Yes" {
-			return fmt.Errorf("Replication on %+v is broken: Slave_IO_Running: %s, Slave_SQL_Running: %s. Please make sure replication runs before using gh-ost.",
+			return fmt.Errorf("Replication on %+v is broken: %s: %s, %s: %s. Please make sure replication runs before using gh-ost.",
 				connectionConfig.Key,
+				ioRunningTerm,
 				slaveIORunning,
+				sqlRunningTerm,
 				slaveSQLRunning,
 			)
 		}
 
 		masterKey = &InstanceKey{
-			Hostname: rowMap.GetString("Master_Host"),
-			Port:     rowMap.GetInt("Master_Port"),
+			Hostname: rowMap.GetString(ReplicaTermFor(dbVersion, "Master_Host")),
+			Port:     rowMap.GetInt(ReplicaTermFor(dbVersion, "Master_Port")),
 		}
 		return nil
 	})
 
 	return masterKey, err
 }
 
-func GetMasterConnectionConfigSafe(connectionConfig *ConnectionConfig, visitedKeys *InstanceKeyMap, allowMasterMaster bool) (masterConfig *ConnectionConfig, err error) {
-	log.Debugf("Looking for master on %+v", connectionConfig.Key)
+func GetMasterConnectionConfigSafe(dbVersion string, connectionConfig *ConnectionConfig, visitedKeys *InstanceKeyMap, allowMasterMaster bool) (masterConfig *ConnectionConfig, err error) {
+	log.Debugf("Looking for %s on %+v", ReplicaTermFor(dbVersion, "master"), connectionConfig.Key)
 
-	masterKey, err := GetMasterKeyFromSlaveStatus(connectionConfig)
+	masterKey, err := GetMasterKeyFromSlaveStatus(dbVersion, connectionConfig)
 	if err != nil {
 		return nil, err
 	}
@@ -131,34 +139,36 @@ func GetMasterConnectionConfigSafe(connectionConfig *ConnectionConfig, visitedKe
 	masterConfig = connectionConfig.Duplicate()
 	masterConfig.Key = *masterKey
 
-	log.Debugf("Master of %+v is %+v", connectionConfig.Key, masterConfig.Key)
+	log.Debugf("%s of %+v is %+v", ReplicaTermFor(dbVersion, "master"), connectionConfig.Key, masterConfig.Key)
 	if visitedKeys.HasKey(masterConfig.Key) {
 		if allowMasterMaster {
 			return connectionConfig, nil
 		}
 		return nil, fmt.Errorf("There seems to be a master-master setup at %+v. This is unsupported. Bailing out", masterConfig.Key)
 	}
 	visitedKeys.AddKey(masterConfig.Key)
-	return GetMasterConnectionConfigSafe(masterConfig, visitedKeys, allowMasterMaster)
+	return GetMasterConnectionConfigSafe(dbVersion, masterConfig, visitedKeys, allowMasterMaster)
 }
 
-func GetReplicationBinlogCoordinates(db *gosql.DB) (readBinlogCoordinates *BinlogCoordinates, executeBinlogCoordinates *BinlogCoordinates, err error) {
-	err = sqlutils.QueryRowsMap(db, `show slave status`, func(m sqlutils.RowMap) error {
+func GetReplicationBinlogCoordinates(dbVersion string, db *gosql.DB) (readBinlogCoordinates *BinlogCoordinates, executeBinlogCoordinates *BinlogCoordinates, err error) {
+	showReplicaStatusQuery := fmt.Sprintf("show %s", ReplicaTermFor(dbVersion, `slave status`))
+	err = sqlutils.QueryRowsMap(db, showReplicaStatusQuery, func(m sqlutils.RowMap) error {
 		readBinlogCoordinates = &BinlogCoordinates{
-			LogFile: m.GetString("Master_Log_File"),
-			LogPos:  m.GetInt64("Read_Master_Log_Pos"),
+			LogFile: m.GetString(ReplicaTermFor(dbVersion, "Master_Log_File")),
+			LogPos:  m.GetInt64(ReplicaTermFor(dbVersion, "Read_Master_Log_Pos")),
 		}
 		executeBinlogCoordinates = &BinlogCoordinates{
-			LogFile: m.GetString("Relay_Master_Log_File"),
-			LogPos:  m.GetInt64("Exec_Master_Log_Pos"),
+			LogFile: m.GetString(ReplicaTermFor(dbVersion, "Relay_Master_Log_File")),
+			LogPos:  m.GetInt64(ReplicaTermFor(dbVersion, "Exec_Master_Log_Pos")),
 		}
 		return nil
 	})
 	return readBinlogCoordinates, executeBinlogCoordinates, err
 }
 
-func GetSelfBinlogCoordinates(db *gosql.DB) (selfBinlogCoordinates *BinlogCoordinates, err error) {
-	err = sqlutils.QueryRowsMap(db, `show master status`, func(m sqlutils.RowMap) error {
+func GetSelfBinlogCoordinates(dbVersion string, db *gosql.DB) (selfBinlogCoordinates *BinlogCoordinates, err error) {
+	binaryLogStatusTerm := ReplicaTermFor(dbVersion, "master status")
+	err = sqlutils.QueryRowsMap(db, fmt.Sprintf("show %s", binaryLogStatusTerm), func(m sqlutils.RowMap) error {
 		selfBinlogCoordinates = &BinlogCoordinates{
 			LogFile: m.GetString("File"),
 			LogPos:  m.GetInt64("Position"),