Merge branch 'master' into touch-postpone-flag-file

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 (
@@ -59,3 +63,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()
 }

localtests/test.sh

@@ -29,6 +29,10 @@ verify_master_and_replica() {
     echo "Cannot verify gh-ost-test-mysql-replica"
     exit 1
   fi
+  if [ "$(gh-ost-test-mysql-replica -e "select @@global.binlog_format" -ss)" != "ROW" ] ; then
+    echo "Expecting test replica to have binlog_format=ROW"
+    exit 1
+  fi
   read replica_host replica_port <<< $(gh-ost-test-mysql-replica -e "select @@hostname, @@port" -ss)
 }
 
@@ -42,6 +46,21 @@ echo_dot() {
   echo -n "."
 }
 
+start_replication() {
+  gh-ost-test-mysql-replica -e "stop slave; start slave;"
+  num_attempts=0
+  while gh-ost-test-mysql-replica -e "show slave status\G" | grep Seconds_Behind_Master | grep -q NULL ; do
+    ((num_attempts=num_attempts+1))
+    if [ $num_attempts -gt 10 ] ; then
+      echo
+      echo "ERROR replication failure"
+      exit 1
+    fi
+    echo_dot
+    sleep 1
+  done
+}
+
 test_single() {
   local test_name
   test_name="$1"
@@ -49,7 +68,7 @@ test_single() {
   echo -n "Testing: $test_name"
 
   echo_dot
-  gh-ost-test-mysql-replica -e "stop slave; start slave; do sleep(1)"
+  start_replication
   echo_dot
   gh-ost-test-mysql-master --default-character-set=utf8mb4 test < $tests_path/$test_name/create.sql
 
@@ -82,7 +101,7 @@ test_single() {
     --table=gh_ost_test \
     --alter='engine=innodb' \
     --exact-rowcount \
-    --switch-to-rbr \
+    --assume-rbr \
     --initially-drop-old-table \
     --initially-drop-ghost-table \
     --throttle-query='select timestampdiff(second, min(last_update), now()) < 5 from _gh_ost_test_ghc' \