Allow gh-ost to modify the server using extra port

Both Percona and Maria allow MySQL to be configured to listen on an extra port when their thread pool is enable.

* https://www.percona.com/doc/percona-server/5.7/performance/threadpool.html
* https://mariadb.com/kb/en/the-mariadb-library/thread-pool-in-mariadb-51-53/

This is valuable because if the table has a lot of traffic (read or write load), gh-ost can end up starving the thread pool as incomming connections are immediately blocked.

By using gh-ost on the extra port, MySQL locking will still behave the same, but MySQL will keep a dedicated thread for each gh-ost connection.

When doing this, it's important to inspect the extra-max-connections variable. Both Percona and Maria default to 1, so gh-ost may easily exceed with its threads.

An example local run using this

```
$ mysql -S /tmp/mysql_sandbox20393.sock -e "select @@global.port, @@global.extra_port"
+---------------+---------------------+
| @@global.port | @@global.extra_port |
+---------------+---------------------+
|         20393 |               30393 |
+---------------+---------------------+

./bin/gh-ost \
--initially-drop-ghost-table \
--initially-drop-old-table \
--assume-rbr \
--port="20395" \
--assume-master-host="127.0.0.1:30393" \
--max-load=Threads_running=25 \
--critical-load=Threads_running=1000 \
--chunk-size=1000 \
--max-lag-millis=1500 \
--user="gh-ost" \
--password="gh-ost" \
--database="test" \
--table="mytable" \
--verbose \
--alter="ADD mynewcol decimal(11,2) DEFAULT 0.0 NOT NULL" \
--exact-rowcount \
--concurrent-rowcount \
--default-retries=120 \
--panic-flag-file=/tmp/ghost.panic.flag \
--postpone-cut-over-flag-file=/tmp/ghost.postpone.flag \
--execute
```

Author: zmoazeni

go/base/utils.go

@@ -11,6 +11,10 @@ import (
 	"regexp"
 	"strings"
 	"time"
+
+	gosql "database/sql"
+	"github.com/github/gh-ost/go/mysql"
+	"github.com/outbrain/golib/log"
 )
 
 var (
@@ -50,3 +54,25 @@ func StringContainsAll(s string, substrings ...string) bool {
 	}
 	return nonEmptyStringsFound
 }
+
+func ValidateConnection(db *gosql.DB, connectionConfig *mysql.ConnectionConfig) (string, error) {
+	query := `select @@global.port, @@global.version`
+	var port, extraPort int
+	var version string
+	if err := db.QueryRow(query).Scan(&port, &version); err != nil {
+		return "", err
+	}
+	extraPortQuery := `select @@global.extra_port`
+	if err := db.QueryRow(extraPortQuery).Scan(&extraPort); err != nil {
+		// swallow this error. not all servers support extra_port
+	}
+
+	if connectionConfig.Key.Port == port || (extraPort > 0 && connectionConfig.Key.Port == extraPort) {
+		log.Infof("connection validated on %+v", connectionConfig.Key)
+		return version, nil
+	} else if extraPort == 0 {
+		return "", fmt.Errorf("Unexpected database port reported: %+v", port)
+	} else {
+		return "", fmt.Errorf("Unexpected database port reported: %+v / extra_port: %+v", port, extraPort)
+	}
+}

go/logic/applier.go

@@ -53,12 +53,14 @@ func (this *Applier) InitDBConnections() (err error) {
 		return err
 	}
 	this.singletonDB.SetMaxOpenConns(1)
-	if err := this.validateConnection(this.db); err != nil {
+	version, err := base.ValidateConnection(this.db, this.connectionConfig)
+	if err != nil {
 		return err
 	}
-	if err := this.validateConnection(this.singletonDB); err != nil {
+	if _, err := base.ValidateConnection(this.singletonDB, this.connectionConfig); err != nil {
 		return err
 	}
+	this.migrationContext.ApplierMySQLVersion = version
 	if err := this.validateAndReadTimeZone(); err != nil {
 		return err
 	}
@@ -74,20 +76,6 @@ func (this *Applier) InitDBConnections() (err error) {
 	return nil
 }
 
-// validateConnection issues a simple can-connect to MySQL
-func (this *Applier) validateConnection(db *gosql.DB) error {
-	query := `select @@global.port, @@global.version`
-	var port int
-	if err := db.QueryRow(query).Scan(&port, &this.migrationContext.ApplierMySQLVersion); err != nil {
-		return err
-	}
-	if port != this.connectionConfig.Key.Port {
-		return fmt.Errorf("Unexpected database port reported: %+v", port)
-	}
-	log.Infof("connection validated on %+v", this.connectionConfig.Key)
-	return nil
-}
-
 // validateAndReadTimeZone potentially reads server time-zone
 func (this *Applier) validateAndReadTimeZone() error {
 	query := `select @@global.time_zone`

go/logic/inspect.go

@@ -195,16 +195,10 @@ func (this *Inspector) validateConnection() error {
 	if len(this.connectionConfig.Password) > mysql.MaxReplicationPasswordLength {
 		return fmt.Errorf("MySQL replication length limited to 32 characters. See https://dev.mysql.com/doc/refman/5.7/en/assigning-passwords.html")
 	}
-	query := `select @@global.port, @@global.version`
-	var port int
-	if err := this.db.QueryRow(query).Scan(&port, &this.migrationContext.InspectorMySQLVersion); err != nil {
-		return err
-	}
-	if port != this.connectionConfig.Key.Port {
-		return fmt.Errorf("Unexpected database port reported: %+v", port)
-	}
-	log.Infof("connection validated on %+v", this.connectionConfig.Key)
-	return nil
+
+	version, err := base.ValidateConnection(this.db, this.connectionConfig)
+	this.migrationContext.InspectorMySQLVersion = version
+	return err
 }
 
 // validateGrants verifies the user by which we're executing has necessary grants

go/logic/streamer.go

@@ -107,7 +107,7 @@ func (this *EventsStreamer) InitDBConnections() (err error) {
 	if this.db, _, err = sqlutils.GetDB(EventsStreamerUri); err != nil {
 		return err
 	}
-	if err := this.validateConnection(); err != nil {
+	if _, err := base.ValidateConnection(this.db, this.connectionConfig); err != nil {
 		return err
 	}
 	if err := this.readCurrentBinlogCoordinates(); err != nil {
@@ -133,20 +133,6 @@ func (this *EventsStreamer) initBinlogReader(binlogCoordinates *mysql.BinlogCoor
 	return nil
 }
 
-// validateConnection issues a simple can-connect to MySQL
-func (this *EventsStreamer) validateConnection() error {
-	query := `select @@global.port`
-	var port int
-	if err := this.db.QueryRow(query).Scan(&port); err != nil {
-		return err
-	}
-	if port != this.connectionConfig.Key.Port {
-		return fmt.Errorf("Unexpected database port reported: %+v", port)
-	}
-	log.Infof("connection validated on %+v", this.connectionConfig.Key)
-	return nil
-}
-
 func (this *EventsStreamer) GetCurrentBinlogCoordinates() *mysql.BinlogCoordinates {
 	return this.binlogReader.GetCurrentBinlogCoordinates()
 }