Use replica status for throttling via control replicas in move-tables mode

Author: danieljoos

go/logic/throttler.go

@@ -209,6 +209,14 @@ func (thlr *Throttler) collectControlReplicasLag() {
 			return lag, err
 		}
 
+		if thlr.migrationContext.IsMoveTablesMode() {
+			dbVersion, err := mysql.GetDBVersion(thlr.migrationContext.Uuid, dbUri)
+			if err != nil {
+				return lag, err
+			}
+			return mysql.GetReplicationLagFromSlaveStatus(dbVersion, db)
+		}
+
 		if err := db.QueryRow(replicationLagQuery).Scan(&heartbeatValue); err != nil {
 			return lag, err
 		}

go/mysql/utils.go

@@ -47,6 +47,8 @@ func (rlg *ReplicationLagResult) HasLag() bool {
 // knownDBs is a DB cache by uri
 var knownDBs map[string]*gosql.DB = make(map[string]*gosql.DB)
 var knownDBsMutex = &sync.Mutex{}
+var knownDBsVersions map[string]string = make(map[string]string)
+var knownDBsVersionsMutex = &sync.Mutex{}
 
 func GetDB(migrationUuid string, mysql_uri string) (db *gosql.DB, exists bool, err error) {
 	cacheKey := migrationUuid + ":" + mysql_uri
@@ -66,6 +68,30 @@ func GetDB(migrationUuid string, mysql_uri string) (db *gosql.DB, exists bool, e
 	return db, exists, nil
 }
 
+// GetDBVersion returns the MySQL version for a given mysql_uri, and caches it for future calls.
+// Uses GetDB to get a connection to the database.
+func GetDBVersion(migrationUuid string, mysql_uri string) (dbVersion string, err error) {
+	cacheKey := migrationUuid + ":" + mysql_uri
+
+	knownDBsVersionsMutex.Lock()
+	defer knownDBsVersionsMutex.Unlock()
+
+	if dbVersion, exists := knownDBsVersions[cacheKey]; exists {
+		return dbVersion, nil
+	}
+
+	db, _, err := GetDB(migrationUuid, mysql_uri)
+	if err != nil {
+		return "", err
+	}
+	var version string
+	if err := db.QueryRow(`select @@global.version`).Scan(&version); err != nil {
+		return "", err
+	}
+	knownDBsVersions[cacheKey] = version
+	return version, nil
+}
+
 // GetReplicationLagFromSlaveStatus returns replication lag for a given db; via SHOW SLAVE STATUS
 func GetReplicationLagFromSlaveStatus(dbVersion string, informationSchemaDb *gosql.DB) (replicationLag time.Duration, err error) {
 	showReplicaStatusQuery := fmt.Sprintf("show %s", ReplicaTermFor(dbVersion, `slave status`))