Merge pull request #1718 from github/move-tables-connection-audit

Introduce a SourcePrimaryConnectionConfig for primary-required move-table ops

Author: zacharysierakowski

go/base/context.go

@@ -277,14 +277,31 @@ type MigrationContext struct {
 
 	// move tables:
 	MoveTables struct {
-		TableNames       []string // List of table names to be moved.
-		TargetHost       string   // Target hostname for the move. This must be a primary/writable host.
-		TargetPort       int      // Target MySQL port for the move.
-		TargetUser       string   // Target username for the move. If not specified, it will default to the source user.
-		TargetPass       string   // Target password for the move. If not specified, it will default to the source password.
-		TargetDatabase   string   // Target database name for the move. If not specified, it will default to the source database name.
+		TableNames     []string // List of table names to be moved.
+		TargetHost     string   // Target hostname for the move. This must be a primary/writable host.
+		TargetPort     int      // Target MySQL port for the move.
+		TargetUser     string   // Target username for the move. If not specified, it will default to the source user.
+		TargetPass     string   // Target password for the move. If not specified, it will default to the source password.
+		TargetDatabase string   // Target database name for the move. If not specified, it will default to the source database name.
+
+		// AllowOnSourcePrimary opts in to running the move-tables read path (schema
+		// inspection, the full row copy, binlog streaming) directly against the
+		// source cluster's primary. By default gh-ost stops early when --host is the
+		// primary, since reading the whole table copy from the primary is the load
+		// move-tables is meant to avoid; the operator should point --host at a replica.
+		AllowOnSourcePrimary bool
+
 		ConnectionConfig *mysql.ConnectionConfig
 
+		// SourcePrimaryConnectionConfig is the detected source-cluster primary. All
+		// source reads (schema inspection, row copy, binlog streaming) go through the
+		// inspector config (InspectorConnectionConfig), which may point at a read
+		// replica to take load off the primary. The cutover RENAME + drain-GTID
+		// capture and the source `__del` DROP must run on a writable primary, so they
+		// use this dedicated config. When the source --host is itself the primary (no
+		// replica topology), detection returns the inspector key and the two coincide.
+		SourcePrimaryConnectionConfig *mysql.ConnectionConfig
+
 		DrainGTID mysql.BinlogCoordinates // Source @@gtid_executed captured immediately after the source RENAME TABLE; the applier drains until it reaches this coordinate (move-tables only).
 	}
 

go/cmd/gh-ost/main.go

@@ -195,6 +195,7 @@ func main() {
 	flag.StringVar(&migrationContext.MoveTables.TargetUser, "target-user", "", "Target MySQL username for --move-tables mode. If not provided, uses the same user as the source connection")
 	flag.StringVar(&migrationContext.MoveTables.TargetPass, "target-password", "", "Target MySQL password for --move-tables mode. If not provided, uses the same password as the source connection")
 	flag.StringVar(&migrationContext.MoveTables.TargetDatabase, "target-database", "", "Target MySQL database name for --move-tables mode. If not provided, uses the same database name as the source connection")
+	flag.BoolVar(&migrationContext.MoveTables.AllowOnSourcePrimary, "allow-on-source-primary", false, "allow --move-tables to read (schema, row copy, binlog) from the source cluster's primary. By default gh-ost stops if --host is the primary; prefer pointing --host at a replica to spare the primary the copy load.")
 
 	flag.CommandLine.SetOutput(os.Stdout)
 	flag.Parse()

go/logic/applier.go

@@ -197,6 +197,13 @@ func (apl *Applier) InitDBConnections() (err error) {
 		if _, err := base.ValidateConnection(apl.moveTablesTargetDB, apl.moveTablesConnectionConfig, apl.migrationContext, apl.name); err != nil {
 			return err
 		}
+		// Fail fast if the move-tables target is not a writable primary. All target
+		// work (table create, row-copy INSERT, checkpoint writes, checkpoint DROP)
+		// requires a writable host; catching read_only here turns a confusing
+		// mid-run write failure into a clear startup error.
+		if err := assertConnectionWritable(apl.moveTablesTargetDB, apl.moveTablesConnectionConfig.Key, "target"); err != nil {
+			return err
+		}
 	}
 	apl.migrationContext.Log.Infof("Applier initiated on %+v, version %+v", apl.connectionConfig.ImpliedKey, apl.migrationContext.ApplierMySQLVersion)
 	return nil

go/logic/migrator.go

@@ -69,3 +69,15 @@ func TestLogMoveTablesRollbackHint_EmitsRenameCommand(t *testing.T) {
 	require.True(t, logger.has("-- rename table `source_db`.`_t_del` to `source_db`.`t`"),
 		"must emit the rename command to roll the source table back")
 }
+
+// TestMoveTablesDropSourceOldTable_NilSourcePrimaryErrors verifies the source
+// `__del` drop fails cleanly (rather than panicking) when the source-primary
+// connection was never initialized. The drop must never silently no-op.
+func TestMoveTablesDropSourceOldTable_NilSourcePrimaryErrors(t *testing.T) {
+	m, _ := newCleanupTestMigrator()
+
+	err := m.dropSourceOldTable()
+
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "source primary connection not initialized")
+}

go/logic/migrator_move_tables_cleanup_test.go

@@ -24,10 +24,10 @@ import (
 
 // -----------------------------------------------------------------------------
 // Pure unit tests - no MySQL. These exercise the orchestration branches that
-// run BEFORE T1's RENAME, so they do not require a real inspector.db. Per the
-// Option A decision in commit 3's plan, the "RENAME was not attempted" check
-// is a proxy assertion: m.inspector is nil, so if T1 were reached the test
-// would panic instead of silently passing.
+// run BEFORE T1's RENAME, so they do not require a real source-primary DB. The
+// "RENAME was not attempted" check is a proxy assertion: m.sourcePrimaryDB is
+// nil, so if T1 were reached the test would fail with the source-primary-not-
+// initialized error instead of the earlier error it asserts on.
 // -----------------------------------------------------------------------------
 
 // TestMoveTablesCutOver_NoopShortCircuits maps to the Noop semantics decision
@@ -55,8 +55,9 @@ func TestMoveTablesCutOver_NoopShortCircuits(t *testing.T) {
 
 // TestMoveTablesCutOver_OnBeforeCutOverHookAbortsBeforeRename maps to T0 in
 // coop_cutover.md section 1.3 ("non-zero return code aborts cutover"). The "aborts
-// BEFORE source DDL" assertion is enforced as a proxy: m.inspector is nil, so
-// if T1 RENAME executed via mgtr.inspector.db, this test would panic.
+// BEFORE source DDL" assertion is enforced as a proxy: m.sourcePrimaryDB is nil,
+// so if T1 RENAME executed it would fail with the source-primary-not-initialized
+// error rather than the asserted hook error.
 func TestMoveTablesCutOver_OnBeforeCutOverHookAbortsBeforeRename(t *testing.T) {
 	var calls []string
 	boom := errors.New("hook says no")
@@ -184,6 +185,67 @@ func TestResumeMoveTablesCutOverFromCheckpointAlreadyDrained(t *testing.T) {
 	}
 }
 
+// TestResolveSourcePrimaryConnectionConfig_AssumeMasterHostnameOverride verifies
+// that --assume-master-host forces the move-tables source primary to the given
+// host, and that --master-user/--master-password override the inherited source
+// credentials. No DB is required: the override branch builds the config purely
+// from the inspector config.
+func TestResolveSourcePrimaryConnectionConfig_AssumeMasterHostnameOverride(t *testing.T) {
+	mc := base.NewMigrationContext()
+	mc.InspectorConnectionConfig.User = "src_user"
+	mc.InspectorConnectionConfig.Password = "src_pass"
+	mc.AssumeMasterHostname = "10.0.0.5:3307"
+	mc.CliMasterUser = "master_user"
+	mc.CliMasterPassword = "master_pass"
+	m := NewMigrator(mc, "test")
+
+	cfg, err := m.resolveSourcePrimaryConnectionConfig("8.0.42")
+	require.NoError(t, err)
+	require.Equal(t, "10.0.0.5", cfg.Key.Hostname)
+	require.Equal(t, 3307, cfg.Key.Port)
+	require.Equal(t, "master_user", cfg.User, "--master-user must override source credentials")
+	require.Equal(t, "master_pass", cfg.Password, "--master-password must override source credentials")
+
+	// Without explicit master credentials, the forced primary inherits the source
+	// (inspector) credentials.
+	mc.CliMasterUser = ""
+	mc.CliMasterPassword = ""
+	cfg, err = m.resolveSourcePrimaryConnectionConfig("8.0.42")
+	require.NoError(t, err)
+	require.Equal(t, "src_user", cfg.User)
+	require.Equal(t, "src_pass", cfg.Password)
+}
+
+// TestValidateMoveTablesSourceReadHost verifies the read-path guard: a replica
+// source passes, a source that resolves to the primary is blocked with a hint to
+// use a replica or --allow-on-source-primary, and the opt-in flag bypasses it.
+func TestValidateMoveTablesSourceReadHost(t *testing.T) {
+	newMigrator := func(srcKey, primaryKey mysql.InstanceKey, allow bool) *Migrator {
+		mc := base.NewMigrationContext()
+		mc.MoveTables.TableNames = []string{"t"}
+		mc.InspectorConnectionConfig.Key = srcKey
+		mc.MoveTables.SourcePrimaryConnectionConfig = &mysql.ConnectionConfig{Key: primaryKey}
+		mc.MoveTables.AllowOnSourcePrimary = allow
+		return NewMigrator(mc, "test")
+	}
+	replica := mysql.InstanceKey{Hostname: "replica.example.com", Port: 3306}
+	primary := mysql.InstanceKey{Hostname: "primary.example.com", Port: 3306}
+
+	t.Run("source is a replica: passes", func(t *testing.T) {
+		require.NoError(t, newMigrator(replica, primary, false).validateMoveTablesSourceReadHost())
+	})
+
+	t.Run("source is the primary: blocked", func(t *testing.T) {
+		err := newMigrator(primary, primary, false).validateMoveTablesSourceReadHost()
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "--allow-on-source-primary")
+	})
+
+	t.Run("source is the primary but opted in: passes", func(t *testing.T) {
+		require.NoError(t, newMigrator(primary, primary, true).validateMoveTablesSourceReadHost())
+	})
+}
+
 // -----------------------------------------------------------------------------
 // Integration tests - real MySQL via testcontainers, exercise T1/T2/T3.
 //
@@ -251,8 +313,10 @@ func (s *MoveTablesCutOverSuite) containingDrainGTID() *mysql.GTIDBinlogCoordina
 }
 
 // buildMigrator wires a Migrator with the test container's *sql.DB pinned to
-// inspector.db and a fresh Applier. initialCoords may be nil for the drain-
-// timeout case.
+// inspector.db and a fresh Applier. The cutover RENAME + drain-GTID capture run
+// on sourcePrimaryDB (a dedicated handle with multiStatements enabled, since
+// T1/T2 are issued as a single multi-statement round trip). initialCoords may be
+// nil for the drain-timeout case.
 func (s *MoveTablesCutOverSuite) buildMigrator(fakeHooks base.Hooks, initialCoords mysql.BinlogCoordinates) (*Migrator, *base.MigrationContext) {
 	ctx := context.Background()
 	connectionConfig, err := getTestConnectionConfig(ctx, s.mysqlContainer)
@@ -261,11 +325,17 @@ func (s *MoveTablesCutOverSuite) buildMigrator(fakeHooks base.Hooks, initialCoor
 	mc := newTestMigrationContext()
 	mc.ApplierConnectionConfig = connectionConfig
 	mc.InspectorConnectionConfig = connectionConfig
+	mc.MoveTables.SourcePrimaryConnectionConfig = connectionConfig
 	mc.SetConnectionConfig("innodb")
 	mc.Hooks = fakeHooks
 
 	m := NewMigrator(mc, "test")
 	m.inspector = &Inspector{db: s.db, migrationContext: mc}
+	// The source primary handle needs multiStatements enabled for the consolidated
+	// T1+T2 (RENAME; SELECT @@global.gtid_executed) round trip.
+	sourcePrimaryDB, _, err := mysql.GetDB(mc.Uuid, connectionConfig.GetDBUri(testMysqlDatabase)+"&multiStatements=true")
+	s.Require().NoError(err)
+	m.sourcePrimaryDB = sourcePrimaryDB
 	m.applier = NewApplier(mc)
 	if initialCoords != nil {
 		m.applier.CurrentCoordinatesMutex.Lock()
@@ -275,6 +345,58 @@ func (s *MoveTablesCutOverSuite) buildMigrator(fakeHooks base.Hooks, initialCoor
 	return m, mc
 }
 
+// TestDropSourceOldTableUsesSourcePrimary verifies the source `__del` rollback
+// handle is dropped through the dedicated source-primary connection. In
+// production the inspector/streamer source connections may be a read replica, so
+// the drop must not route through them.
+func (s *MoveTablesCutOverSuite) TestDropSourceOldTableUsesSourcePrimary() {
+	ctx := context.Background()
+	_, err := s.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY)", getTestOldTableName()))
+	s.Require().NoError(err)
+
+	var calls []string
+	fakeHooks := &recordingHooks{name: "fake", calls: &calls}
+	m, _ := s.buildMigrator(fakeHooks, s.containingDrainGTID())
+
+	s.Require().NoError(m.dropSourceOldTable())
+
+	var name string
+	err = s.db.QueryRow(fmt.Sprintf("SHOW TABLES IN %s LIKE '_%s_del'",
+		testMysqlDatabase, testMysqlTableName)).Scan(&name)
+	s.Require().ErrorIs(err, gosql.ErrNoRows, "source __del handle must be dropped via the source primary")
+}
+
+// TestResolveSourcePrimaryFallsBackToInspectorWhenNoReplica verifies the
+// graceful fallback: when the source --host has no upstream primary (the
+// standalone test container), master detection returns the inspector connection
+// config, so source reads and cutover writes share the one available host.
+func (s *MoveTablesCutOverSuite) TestResolveSourcePrimaryFallsBackToInspectorWhenNoReplica() {
+	var calls []string
+	m, mc := s.buildMigrator(&recordingHooks{name: "fake", calls: &calls}, nil)
+
+	cfg, err := m.resolveSourcePrimaryConnectionConfig("8.0.42")
+	s.Require().NoError(err)
+	s.Require().Equal(mc.InspectorConnectionConfig.Key.Hostname, cfg.Key.Hostname)
+	s.Require().Equal(mc.InspectorConnectionConfig.Key.Port, cfg.Key.Port)
+}
+
+// TestAssertConnectionWritableRejectsReadOnly verifies the startup writability
+// gate: a writable primary passes, and a read_only server is rejected with a
+// clear error. The same helper guards both the source primary and the target.
+func (s *MoveTablesCutOverSuite) TestAssertConnectionWritableRejectsReadOnly() {
+	key := mysql.InstanceKey{Hostname: "test-host", Port: 3306}
+	s.Require().NoError(assertConnectionWritable(s.db, key, "source primary"),
+		"a writable primary must pass the gate")
+
+	_, err := s.db.Exec("SET GLOBAL read_only = ON")
+	s.Require().NoError(err)
+	defer func() { _, _ = s.db.Exec("SET GLOBAL read_only = OFF") }()
+
+	err = assertConnectionWritable(s.db, key, "source primary")
+	s.Require().Error(err)
+	s.Require().Contains(err.Error(), "read_only")
+}
+
 // TestHappyPath drives the full T0-T6 protocol against the test container.
 // Asserts hook ordering (T0 then T5), T4 flag set, and the source-side rename.
 // Maps to acceptance criterion #8209 "RENAME executes; drain completes;
@@ -322,7 +444,7 @@ func (s *MoveTablesCutOverSuite) TestRenameFailurePropagates() {
 
 	err := m.moveTablesCutOver()
 	s.Require().Error(err)
-	s.Require().Contains(err.Error(), "RENAME failed")
+	s.Require().Contains(err.Error(), "source RENAME + drain GTID capture failed")
 
 	s.Require().Equal(int64(0), atomic.LoadInt64(&mc.CutOverCompleteFlag),
 		"post-state: RENAME failure must leave CutOverCompleteFlag unset")

go/logic/migrator_move_tables_cutover_test.go

@@ -15,7 +15,6 @@ import (
 	"github.com/github/gh-ost/go/base"
 	"github.com/github/gh-ost/go/binlog"
 	"github.com/github/gh-ost/go/mysql"
-	"github.com/github/gh-ost/go/sql"
 
 	gomysql "github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/openark/golib/sqlutils"
@@ -240,32 +239,6 @@ func (es *EventsStreamer) Close() (err error) {
 	return err
 }
 
-// DropSourceOldTable drops the source "__del" table in move-tables mode. The
-// __del table is the post-cutover rollback handle on the source cluster; it is
-// only dropped after a successful run when --ok-to-drop-table is set.
-// The applier's dropTable targets the move-tables target cluster,
-// so the source-side drop is owned by the streamer: its `db`
-// handle uses InspectorConnectionConfig (the source) and stays open in both the
-// normal and the cutover-resume paths (Close() only closes the binlog reader,
-// not `db`).
-func (es *EventsStreamer) DropSourceOldTable() error {
-	databaseName := es.migrationContext.DatabaseName
-	tableName := es.migrationContext.GetOldTableName()
-	query := fmt.Sprintf(`drop /* gh-ost */ table if exists %s.%s`,
-		sql.EscapeName(databaseName),
-		sql.EscapeName(tableName),
-	)
-	es.migrationContext.Log.Infof("Dropping source table %s.%s",
-		sql.EscapeName(databaseName),
-		sql.EscapeName(tableName),
-	)
-	if _, err := sqlutils.ExecNoPrepare(es.db, query); err != nil {
-		return err
-	}
-	es.migrationContext.Log.Infof("Source table dropped")
-	return nil
-}
-
 func (es *EventsStreamer) Teardown() {
 	es.db.Close()
 }

go/logic/streamer.go

@@ -181,6 +181,7 @@ build_ghost_command() {
     --password=opensesame \
     --host=$source_replica_host \
     --port=$source_replica_port \
+    --assume-master-host=${source_master_host}:${source_master_port} \
     --database=$database \
     --target-user=root \
     --target-password=opensesame \

localtests/move-tables-test.sh

@@ -24,14 +24,9 @@ script/build --cli
 
 Run gh-ost to move tables:
 ```bash
-./script/build --cli; ./bin/gh-ost --move-tables=gh_ost_test --host=localhost --port=3307 --user root --password opensesame --database=gh_ost_test_db --target-host=localhost --target-port=3309 --target-user root --target-password opensesame --target-database=gh_ost_test_db --postpone-cut-over-flag-file=/tmp/ghost-move-tables.postpone.flag --execute --verbose --checkpoint --checkpoint-seconds 10
+./script/build --cli; ./bin/gh-ost --move-tables=gh_ost_test --host=localhost --port=3308 --user root --password opensesame --database=gh_ost_test_db --target-host=localhost --target-port=3309 --target-user root --target-password opensesame --target-database=gh_ost_test_db --postpone-cut-over-flag-file=/tmp/ghost-move-tables.postpone.flag --execute --verbose --checkpoint --checkpoint-seconds 10 --initially-drop-socket-file
 ```
 
-Note: replicas in this local topology are configured with `read_only=ON` and
-`super_read_only=ON`. If you point `--host` at `mysql-source-replica` (3308),
-the cutover `RENAME TABLE` step will fail by design. Use source primary (3307)
-as the inspected host when you want cutover to rename on source.
-
 Start continuous inserts against the source.
 ```bash
 script/move-tables/insert-source-primary-loop

script/move-tables/README.md

@@ -0,0 +1,17 @@
+#!/bin/bash
+
+set -euo pipefail
+
+GH_OST_ROOT=$(git rev-parse --show-toplevel)
+SCRIPT_PATH="${GH_OST_ROOT}/script/move-tables"
+DATABASE_NAME="${GH_OST_TEST_DB:-gh_ost_test_db}"
+
+# Reset source table state regardless of whether cutover renamed it.
+${SCRIPT_PATH}/mysql-source-primary -D "${DATABASE_NAME}" -e "DROP TABLE IF EXISTS _gh_ost_test_del, gh_ost_test;"
+
+# Recreate and seed source table data, same fixture as setup uses.
+${SCRIPT_PATH}/mysql-source-primary -D "${DATABASE_NAME}" < "${GH_OST_ROOT}/localtests/move-tables/create.sql"
+
+${SCRIPT_PATH}/mysql-target-primary -D "${DATABASE_NAME}" -e "DROP TABLE IF EXISTS gh_ost_test, _gh_ost_test_ghk;"
+
+echo "Reset source and target tables in ${DATABASE_NAME}"