Fix data races and enable race detector in CI

go/binlog/gomysql_reader.go

@@ -86,13 +86,13 @@ func (gmr *GoMySQLReader) ConnectBinlogStreamer(coordinates mysql.BinlogCoordina
 
 	gmr.currentCoordinatesMutex.Lock()
 	defer gmr.currentCoordinatesMutex.Unlock()
-	gmr.currentCoordinates = coordinates
+	gmr.currentCoordinates = coordinates.Clone()
 	gmr.migrationContext.Log.Infof("Connecting binlog streamer at %+v", coordinates)
 
 	// Start sync with specified GTID set or binlog file and position
 	if gmr.migrationContext.UseGTIDs {
 		coords := coordinates.(*mysql.GTIDBinlogCoordinates)
-		gmr.binlogStreamer, err = gmr.binlogSyncer.StartSyncGTID(coords.GTIDSet)
+		gmr.binlogStreamer, err = gmr.binlogSyncer.StartSyncGTID(coords.GTIDSet.Clone())
 	} else {
 		coords := gmr.currentCoordinates.(*mysql.FileBinlogCoordinates)
 		gmr.binlogStreamer, err = gmr.binlogSyncer.StartSync(gomysql.Position{

go/logic/migrator.go

@@ -87,6 +87,7 @@ type Migrator struct {
 	throttler        *Throttler
 	hooksExecutor    base.Hooks
 	migrationContext *base.MigrationContext
+	statusWriter     io.Writer
 
 	firstThrottlingCollected   chan bool
 	ghostTableMigrated         chan bool
@@ -112,6 +113,7 @@ func NewMigrator(context *base.MigrationContext, appVersion string) *Migrator {
 		appVersion:               appVersion,
 		hooksExecutor:            hooks,
 		migrationContext:         context,
+		statusWriter:             os.Stdout,
 		parser:                   sql.NewAlterTableParser(),
 		ghostTableMigrated:       make(chan bool),
 		firstThrottlingCollected: make(chan bool, 3),
@@ -1481,7 +1483,7 @@ func (mgtr *Migrator) printStatus(rule PrintStatusRule, snap migrationProgressSn
 	if rule == NoPrintStatusRule {
 		return
 	}
-	writers = append(writers, os.Stdout)
+	writers = append(writers, mgtr.statusWriter)
 
 	// Before status, let's see if we should print a nice reminder for what exactly we're doing here.
 	if mgtr.shouldPrintMigrationStatusHint(rule, snap.elapsedSeconds) {

go/logic/migrator_test.go

@@ -35,6 +35,25 @@ import (
 	"github.com/testcontainers/testcontainers-go"
 )
 
+type buffer struct {
+	mu sync.Mutex
+	bytes.Buffer
+}
+
+func (buf *buffer) Write(data []byte) (int, error) {
+	buf.mu.Lock()
+	defer buf.mu.Unlock()
+
+	return buf.Buffer.Write(data)
+}
+
+func (buf *buffer) String() string {
+	buf.mu.Lock()
+	defer buf.mu.Unlock()
+
+	return buf.Buffer.String()
+}
+
 func TestMigratorOnChangelogEvent(t *testing.T) {
 	migrationContext := base.NewMigrationContext()
 	migrator := NewMigrator(migrationContext, "1.2.3")
@@ -824,14 +843,6 @@ echo "[gh-ost-on-batch-copy-retry]: Done, exiting..."
 	err = os.WriteFile(hookScript, []byte(hookContent), 0755)
 	suite.Require().NoError(err)
 
-	origStdout := os.Stdout
-	origStderr := os.Stderr
-
-	rOut, wOut, _ := os.Pipe()
-	rErr, wErr, _ := os.Pipe()
-	os.Stdout = wOut
-	os.Stderr = wErr
-
 	connectionConfig, err := getTestConnectionConfig(ctx, suite.mysqlContainer)
 	suite.Require().NoError(err)
 
@@ -854,19 +865,15 @@ echo "[gh-ost-on-batch-copy-retry]: Done, exiting..."
 	migrationContext.ServeSocketFile = "/tmp/gh-ost.sock"
 
 	migrator := NewMigrator(migrationContext, "0.0.0")
+	hooksExecutor, ok := migrator.hooksExecutor.(*HooksExecutor)
+	suite.Require().True(ok)
+	var bufOut, bufErr buffer
+	migrator.statusWriter = &bufOut
+	hooksExecutor.writer = &bufErr
 
 	err = migrator.Migrate()
 	suite.Require().NoError(err)
 
-	wOut.Close()
-	wErr.Close()
-	os.Stdout = origStdout
-	os.Stderr = origStderr
-
-	var bufOut, bufErr bytes.Buffer
-	io.Copy(&bufOut, rOut)
-	io.Copy(&bufErr, rErr)
-
 	outStr := bufOut.String()
 	errStr := bufErr.String()
 

script/test

@@ -14,4 +14,4 @@ script/build
 cd .gopath/src/github.com/github/gh-ost
 
 echo "Running unit tests"
-go test "$@" -v -covermode=atomic ./go/...
+go test "$@" -v -race -covermode=atomic ./go/...