Refine wait_timeout flag to be cut-over only

Signed-off-by: Tim Vaillancourt <[email protected]>

Author: timvaillancourt

go/base/context.go

@@ -135,6 +135,7 @@ type MigrationContext struct {
 	CriticalLoadHibernateSeconds        int64
 	PostponeCutOverFlagFile             string
 	CutOverLockTimeoutSeconds           int64
+	CutOverIdleTimeoutSeconds           int64
 	CutOverExponentialBackoff           bool
 	ExponentialBackoffMaxInterval       int64
 	ForceNamedCutOverCommand            bool
@@ -164,6 +165,7 @@ type MigrationContext struct {
 	Hostname                               string
 	AssumeMasterHostname                   string
 	ApplierTimeZone                        string
+	ApplierWaitTimeout                     int64
 	TableEngine                            string
 	RowsEstimate                           int64
 	RowsDeltaEstimate                      int64
@@ -276,6 +278,7 @@ func NewMigrationContext() *MigrationContext {
 		ApplierConnectionConfig:             mysql.NewConnectionConfig(),
 		MaxLagMillisecondsThrottleThreshold: 1500,
 		CutOverLockTimeoutSeconds:           3,
+		CutOverIdleTimeoutSeconds:           6,
 		DMLBatchSize:                        10,
 		etaNanoseonds:                       ETAUnknown,
 		maxLoad:                             NewLoadMap(),

go/cmd/gh-ost/main.go

@@ -51,7 +51,6 @@ func main() {
 	flag.StringVar(&migrationContext.AssumeMasterHostname, "assume-master-host", "", "(optional) explicitly tell gh-ost the identity of the master. Format: some.host.com[:port] This is useful in master-master setups where you wish to pick an explicit master, or in a tungsten-replicator where gh-ost is unable to determine the master")
 	flag.IntVar(&migrationContext.InspectorConnectionConfig.Key.Port, "port", 3306, "MySQL port (preferably a replica, not the master)")
 	flag.Float64Var(&migrationContext.InspectorConnectionConfig.Timeout, "mysql-timeout", 0.0, "Connect, read and write timeout for MySQL")
-	flag.Float64Var(&migrationContext.InspectorConnectionConfig.WaitTimeout, "mysql-wait-timeout", 0.0, "wait_timeout for MySQL sessions")
 	flag.StringVar(&migrationContext.CliUser, "user", "", "MySQL user")
 	flag.StringVar(&migrationContext.CliPassword, "password", "", "MySQL password")
 	flag.StringVar(&migrationContext.CliMasterUser, "master-user", "", "MySQL user on master, if different from that on replica. Requires --assume-master-host")
@@ -109,6 +108,7 @@ func main() {
 	dmlBatchSize := flag.Int64("dml-batch-size", 10, "batch size for DML events to apply in a single transaction (range 1-100)")
 	defaultRetries := flag.Int64("default-retries", 60, "Default number of retries for various operations before panicking")
 	cutOverLockTimeoutSeconds := flag.Int64("cut-over-lock-timeout-seconds", 3, "Max number of seconds to hold locks on tables while attempting to cut-over (retry attempted when lock exceeds timeout)")
+	flag.Int64Var(&migrationContext.CutOverIdleTimeoutSeconds, "cut-over-idle-timeout-seconds", *cutOverLockTimeoutSeconds*2, "the idle timeout in seconds for the cut-over operation. The MySQL session performing the cut-over uses this as a wait_timeout.")
 	niceRatio := flag.Float64("nice-ratio", 0, "force being 'nice', imply sleep time per chunk time; range: [0.0..100.0]. Example values: 0 is aggressive. 1: for every 1ms spent copying rows, sleep additional 1ms (effectively doubling runtime); 0.7: for every 10ms spend in a rowcopy chunk, spend 7ms sleeping immediately after")
 
 	maxLagMillis := flag.Int64("max-lag-millis", 1500, "replication lag at which to throttle operation")

go/logic/applier.go

@@ -89,7 +89,7 @@ func (this *Applier) InitDBConnections() (err error) {
 		return err
 	}
 	this.migrationContext.ApplierMySQLVersion = version
-	if err := this.validateAndReadTimeZone(); err != nil {
+	if err := this.validateAndReadGlobalVariables(); err != nil {
 		return err
 	}
 	if !this.migrationContext.AliyunRDS && !this.migrationContext.GoogleCloudPlatform && !this.migrationContext.AzureMySQL {
@@ -106,10 +106,13 @@ func (this *Applier) InitDBConnections() (err error) {
 	return nil
 }
 
-// validateAndReadTimeZone potentially reads server time-zone
-func (this *Applier) validateAndReadTimeZone() error {
-	query := `select /* gh-ost */ @@global.time_zone`
-	if err := this.db.QueryRow(query).Scan(&this.migrationContext.ApplierTimeZone); err != nil {
+// validateAndReadGlobalVariables potentially reads server global variables, such as the time_zone and wait_timeout.
+func (this *Applier) validateAndReadGlobalVariables() error {
+	query := `select /* gh-ost */ @@global.time_zone, @@global.wait_timeout`
+	if err := this.db.QueryRow(query).Scan(
+		&this.migrationContext.ApplierTimeZone,
+		&this.migrationContext.ApplierWaitTimeout,
+	); err != nil {
 		return err
 	}
 
@@ -974,6 +977,22 @@ func (this *Applier) AtomicCutOverMagicLock(sessionIdChan chan int64, tableLocke
 		return err
 	}
 
+	this.migrationContext.Log.Infof("Setting cut-over idle timeout as %d seconds", this.migrationContext.CutOverIdleTimeoutSeconds)
+	query = fmt.Sprintf(`set /* gh-ost */ session wait_timeout:=%d`, this.migrationContext.CutOverIdleTimeoutSeconds)
+	if _, err := tx.Exec(query); err != nil {
+		tableLocked <- err
+		return err
+	}
+	defer func() {
+		this.migrationContext.Log.Infof("Restoring applier idle timeout as %d seconds", this.migrationContext.ApplierWaitTimeout)
+		query = fmt.Sprintf(`set /* gh-ost */ session wait_timeout:=%d`, this.migrationContext.ApplierWaitTimeout)
+		if _, err := sqlutils.ExecNoPrepare(this.db, query); err != nil {
+			this.migrationContext.Log.Errorf("Failed to restore applier wait_timeout to %d seconds: %v",
+				this.migrationContext.ApplierWaitTimeout, err,
+			)
+		}
+	}()
+
 	if err := this.CreateAtomicCutOverSentryTable(); err != nil {
 		tableLocked <- err
 		return err

go/mysql/connection.go

@@ -31,7 +31,6 @@ type ConnectionConfig struct {
 	Timeout              float64
 	TransactionIsolation string
 	Charset              string
-	WaitTimeout          float64
 }
 
 func NewConnectionConfig() *ConnectionConfig {
@@ -52,7 +51,6 @@ func (this *ConnectionConfig) DuplicateCredentials(key InstanceKey) *ConnectionC
 		Timeout:              this.Timeout,
 		TransactionIsolation: this.TransactionIsolation,
 		Charset:              this.Charset,
-		WaitTimeout:          this.WaitTimeout,
 	}
 	config.ImpliedKey = &config.Key
 	return config
@@ -141,9 +139,6 @@ func (this *ConnectionConfig) GetDBUri(databaseName string) string {
 		fmt.Sprintf("readTimeout=%fs", this.Timeout),
 		fmt.Sprintf("writeTimeout=%fs", this.Timeout),
 	}
-	if this.WaitTimeout > 0 {
-		connectionParams = append(connectionParams, fmt.Sprintf("wait_timeout=%fs", this.WaitTimeout))
-	}
 
 	return fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?%s", this.User, this.Password, hostname, this.Key.Port, databaseName, strings.Join(connectionParams, "&"))
 }

go/mysql/connection_test.go

@@ -82,7 +82,6 @@ func TestGetDBUri(t *testing.T) {
 	c.User = "gromit"
 	c.Password = "penguin"
 	c.Timeout = 1.2345
-	c.WaitTimeout = 0 // should be ignored
 	c.TransactionIsolation = transactionIsolation
 	c.Charset = "utf8mb4,utf8,latin1"
 
@@ -96,11 +95,10 @@ func TestGetDBUriWithTLSSetup(t *testing.T) {
 	c.User = "gromit"
 	c.Password = "penguin"
 	c.Timeout = 1.2345
-	c.WaitTimeout = 60
 	c.tlsConfig = &tls.Config{}
 	c.TransactionIsolation = transactionIsolation
 	c.Charset = "utf8mb4_general_ci,utf8_general_ci,latin1"
 
 	uri := c.GetDBUri("test")
-	test.S(t).ExpectEquals(uri, `gromit:penguin@tcp(myhost:3306)/test?autocommit=true&interpolateParams=true&charset=utf8mb4_general_ci,utf8_general_ci,latin1&tls=ghost&transaction_isolation="REPEATABLE-READ"&timeout=1.234500s&readTimeout=1.234500s&writeTimeout=1.234500s&wait_timeout=60.000000s`)
+	test.S(t).ExpectEquals(uri, `gromit:penguin@tcp(myhost:3306)/test?autocommit=true&interpolateParams=true&charset=utf8mb4_general_ci,utf8_general_ci,latin1&tls=ghost&transaction_isolation="REPEATABLE-READ"&timeout=1.234500s&readTimeout=1.234500s&writeTimeout=1.234500s`)
 }