Merge branch 'master' into fix-infinite-cutover-loop

Author: Shlomi Noach

doc/command-line-flags.md

@@ -130,3 +130,7 @@ See `approve-renamed-columns`
 ### test-on-replica
 
 Issue the migration on a replica; do not modify data on master. Useful for validating, testing and benchmarking. See [testing-on-replica](testing-on-replica.md)
+
+### timestamp-old-table
+
+Makes the _old_ table include a timestamp value. The _old_ table is what the original table is renamed to at the end of a successful migration. For example, if the table is `gh_ost_test`, then the _old_ table would normally be `_gh_ost_test_del`. With `--timestamp-old-table` it would be, for example, `_gh_ost_test_20170221103147_del`.

go/base/context.go

@@ -121,6 +121,7 @@ type MigrationContext struct {
 	OkToDropTable                bool
 	InitiallyDropOldTable        bool
 	InitiallyDropGhostTable      bool
+	TimestampOldTable            bool // Should old table name include a timestamp
 	CutOverType                  CutOver
 	ReplicaServerId              uint
 
@@ -234,11 +235,12 @@ func (this *MigrationContext) GetGhostTableName() string {
 
 // GetOldTableName generates the name of the "old" table, into which the original table is renamed.
 func (this *MigrationContext) GetOldTableName() string {
-	if this.TestOnReplica {
-		return fmt.Sprintf("_%s_ght", this.OriginalTableName)
-	}
-	if this.MigrateOnReplica {
-		return fmt.Sprintf("_%s_ghr", this.OriginalTableName)
+	if this.TimestampOldTable {
+		t := this.StartTime
+		timestamp := fmt.Sprintf("%d%02d%02d%02d%02d%02d",
+			t.Year(), t.Month(), t.Day(),
+			t.Hour(), t.Minute(), t.Second())
+		return fmt.Sprintf("_%s_%s_del", this.OriginalTableName, timestamp)
 	}
 	return fmt.Sprintf("_%s_del", this.OriginalTableName)
 }

go/binlog/gomysql_reader.go

@@ -16,6 +16,7 @@ import (
 	"github.com/outbrain/golib/log"
 	gomysql "github.com/siddontang/go-mysql/mysql"
 	"github.com/siddontang/go-mysql/replication"
+	"golang.org/x/net/context"
 )
 
 type GoMySQLReader struct {
@@ -39,7 +40,16 @@ func NewGoMySQLReader(connectionConfig *mysql.ConnectionConfig) (binlogReader *G
 	}
 
 	serverId := uint32(binlogReader.MigrationContext.ReplicaServerId)
-	binlogReader.binlogSyncer = replication.NewBinlogSyncer(serverId, "mysql")
+
+	binlogSyncerConfig := &replication.BinlogSyncerConfig{
+		ServerID: serverId,
+		Flavor:   "mysql",
+		Host:     connectionConfig.Key.Hostname,
+		Port:     uint16(connectionConfig.Key.Port),
+		User:     connectionConfig.User,
+		Password: connectionConfig.Password,
+	}
+	binlogReader.binlogSyncer = replication.NewBinlogSyncer(binlogSyncerConfig)
 
 	return binlogReader, err
 }
@@ -49,10 +59,6 @@ func (this *GoMySQLReader) ConnectBinlogStreamer(coordinates mysql.BinlogCoordin
 	if coordinates.IsEmpty() {
 		return log.Errorf("Emptry coordinates at ConnectBinlogStreamer()")
 	}
-	log.Infof("Registering replica at %+v:%+v", this.connectionConfig.Key.Hostname, uint16(this.connectionConfig.Key.Port))
-	if err := this.binlogSyncer.RegisterSlave(this.connectionConfig.Key.Hostname, uint16(this.connectionConfig.Key.Port), this.connectionConfig.User, this.connectionConfig.Password); err != nil {
-		return err
-	}
 
 	this.currentCoordinates = coordinates
 	log.Infof("Connecting binlog streamer at %+v", this.currentCoordinates)
@@ -126,7 +132,7 @@ func (this *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesCha
 		if canStopStreaming() {
 			break
 		}
-		ev, err := this.binlogStreamer.GetEvent()
+		ev, err := this.binlogStreamer.GetEvent(context.Background())
 		if err != nil {
 			return err
 		}

go/cmd/gh-ost/main.go

@@ -77,6 +77,7 @@ func main() {
 	flag.BoolVar(&migrationContext.OkToDropTable, "ok-to-drop-table", false, "Shall the tool drop the old table at end of operation. DROPping tables can be a long locking operation, which is why I'm not doing it by default. I'm an online tool, yes?")
 	flag.BoolVar(&migrationContext.InitiallyDropOldTable, "initially-drop-old-table", false, "Drop a possibly existing OLD table (remains from a previous run?) before beginning operation. Default is to panic and abort if such table exists")
 	flag.BoolVar(&migrationContext.InitiallyDropGhostTable, "initially-drop-ghost-table", false, "Drop a possibly existing Ghost table (remains from a previous run?) before beginning operation. Default is to panic and abort if such table exists")
+	flag.BoolVar(&migrationContext.TimestampOldTable, "timestamp-old-table", false, "Use a timestamp in old table name. This makes old table names unique and non conflicting cross migrations")
 	cutOver := flag.String("cut-over", "atomic", "choose cut-over type (default|atomic, two-step)")
 	flag.BoolVar(&migrationContext.ForceNamedCutOverCommand, "force-named-cut-over", false, "When true, the 'unpostpone|cut-over' interactive command must name the migrated table")
 

go/logic/applier.go

@@ -142,6 +142,10 @@ func (this *Applier) ValidateOrDropExistingTables() error {
 			return err
 		}
 	}
+	if len(this.migrationContext.GetOldTableName()) > mysql.MaxTableNameLength {
+		log.Fatalf("--timestamp-old-table defined, but resulting table name (%s) is too long (only %d characters allowed)", this.migrationContext.GetOldTableName(), mysql.MaxTableNameLength)
+	}
+
 	if this.tableExists(this.migrationContext.GetOldTableName()) {
 		return fmt.Errorf("Table %s already exists. Panicking. Use --initially-drop-old-table to force dropping it, though I really prefer that you drop it or rename it away", sql.EscapeName(this.migrationContext.GetOldTableName()))
 	}

go/logic/inspect.go

@@ -348,7 +348,7 @@ func (this *Inspector) validateLogSlaveUpdates() error {
 	}
 
 	if this.migrationContext.IsTungsten {
-		log.Warning("log_slave_updates not found on %s:%d, but --tungsten provided, so I'm proceeding", this.connectionConfig.Key.Hostname, this.connectionConfig.Key.Port)
+		log.Warningf("log_slave_updates not found on %s:%d, but --tungsten provided, so I'm proceeding", this.connectionConfig.Key.Hostname, this.connectionConfig.Key.Port)
 		return nil
 	}
 
@@ -357,7 +357,7 @@ func (this *Inspector) validateLogSlaveUpdates() error {
 	}
 
 	if this.migrationContext.InspectorIsAlsoApplier() {
-		log.Warning("log_slave_updates not found on %s:%d, but executing directly on master, so I'm proceeeding", this.connectionConfig.Key.Hostname, this.connectionConfig.Key.Port)
+		log.Warningf("log_slave_updates not found on %s:%d, but executing directly on master, so I'm proceeeding", this.connectionConfig.Key.Hostname, this.connectionConfig.Key.Port)
 		return nil
 	}
 

go/logic/migrator.go

@@ -1030,11 +1030,13 @@ func (this *Migrator) iterateChunks() error {
 	for {
 		if atomic.LoadInt64(&this.rowCopyCompleteFlag) == 1 {
 			// Done
+			// There's another such check down the line
 			return nil
 		}
 		copyRowsFunc := func() error {
 			if atomic.LoadInt64(&this.rowCopyCompleteFlag) == 1 {
-				// Done
+				// Done.
+				// There's another such check down the line
 				return nil
 			}
 			hasFurtherRange, err := this.applier.CalculateNextIterationRangeEndValues()
@@ -1046,6 +1048,17 @@ func (this *Migrator) iterateChunks() error {
 			}
 			// Copy task:
 			applyCopyRowsFunc := func() error {
+				if atomic.LoadInt64(&this.rowCopyCompleteFlag) == 1 {
+					// No need for more writes.
+					// This is the de-facto place where we avoid writing in the event of completed cut-over.
+					// There could _still_ be a race condition, but that's as close as we can get.
+					// What about the race condition? Well, there's actually no data integrity issue.
+					// when rowCopyCompleteFlag==1 that means **guaranteed** all necessary rows have been copied.
+					// But some are still then collected at the binary log, and these are the ones we're trying to
+					// not apply here. If the race condition wins over us, then we just attempt to apply onto the
+					// _ghost_ table, which no longer exists. So, bothering error messages and all, but no damage.
+					return nil
+				}
 				_, rowsAffected, _, err := this.applier.ApplyIterationInsertQuery()
 				if err != nil {
 					return terminateRowIteration(err)

go/mysql/utils.go

@@ -16,6 +16,8 @@ import (
 	"github.com/outbrain/golib/sqlutils"
 )
 
+const MaxTableNameLength = 64
+
 type ReplicationLagResult struct {
 	Key InstanceKey
 	Lag time.Duration