Add move-tables environment variables to hooks

Author: danieljoos

doc/hooks.md

@@ -73,15 +73,20 @@ The following variables are available on all hooks:
 - `GH_OST_MIGRATED_HOST`
 - `GH_OST_INSPECTED_HOST`
 - `GH_OST_EXECUTING_HOST`
+- `GH_OST_TARGET_HOST` - the target/applier hostname (in `--move-tables`, this is the target cluster host)
 - `GH_OST_HOOKS_HINT` - copy of `--hooks-hint` value
 - `GH_OST_HOOKS_HINT_OWNER` - copy of `--hooks-hint-owner` value
 - `GH_OST_HOOKS_HINT_TOKEN` - copy of `--hooks-hint-token` value
 - `GH_OST_DRY_RUN` - whether or not the `gh-ost` run is a dry run
 - `GH_OST_REVERT` - whether or not `gh-ost` is running in revert mode
+- `GH_OST_MOVE_TABLES` - whether or not `gh-ost` is running in `--move-tables` mode
+- `GH_OST_TARGET_DATABASE_NAME` - operation target database name (in `--move-tables`, this is the explicit target database)
+- `GH_OST_TARGET_TABLE_NAME` - operation target table name (mode-dependent)
 
-The following variable are available on particular hooks:
+The following variables are available on particular hooks:
 
 - `GH_OST_INSTANT_DDL` is only available in `gh-ost-on-success`. The value is `true` if instant DDL was successful, and `false` if it was not.
+- `GH_OST_DRAIN_GTID` is only available in `gh-ost-on-success` and only in `--move-tables` mode. It contains the source `@@gtid_executed` captured immediately after the source `RENAME TABLE` (cutover) and represents the drain target the applier waits for.
 - `GH_OST_COMMAND` is only available in `gh-ost-on-interactive-command`
 - `GH_OST_STATUS` is only available in `gh-ost-on-status`
 - `GH_OST_LAST_BATCH_COPY_ERROR` is only available in `gh-ost-on-batch-copy-retry`

go/base/context.go

@@ -283,6 +283,8 @@ type MigrationContext struct {
 		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.
 		ConnectionConfig *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).
 	}
 
 	Log Logger
@@ -491,6 +493,18 @@ func (mctx *MigrationContext) GetInspectorHostname() string {
 	return mctx.InspectorConnectionConfig.ImpliedKey.Hostname
 }
 
+// GetTargetHostname is a safe access method to the target hostname.
+// In move-tables mode, this is the hostname of the target database,
+// otherwise it's the same as the applier hostname.
+func (mctx *MigrationContext) GetTargetHostname() string {
+	if mctx.IsMoveTablesMode() &&
+		mctx.MoveTables.ConnectionConfig != nil &&
+		mctx.MoveTables.ConnectionConfig.ImpliedKey != nil {
+		return mctx.MoveTables.ConnectionConfig.ImpliedKey.Hostname
+	}
+	return mctx.GetApplierHostname()
+}
+
 // InspectorIsAlsoApplier is `true` when the both inspector and applier are the
 // same database instance. This would be true when running directly on master or when
 // testing on replica.

go/logic/hooks.go

@@ -233,6 +233,7 @@ func (he *HooksExecutor) applyEnvironmentVariables(extraVariables ...string) []s
 	env = append(env, fmt.Sprintf("GH_OST_MIGRATED_HOST=%s", he.migrationContext.GetApplierHostname()))
 	env = append(env, fmt.Sprintf("GH_OST_INSPECTED_HOST=%s", he.migrationContext.GetInspectorHostname()))
 	env = append(env, fmt.Sprintf("GH_OST_EXECUTING_HOST=%s", he.migrationContext.Hostname))
+	env = append(env, fmt.Sprintf("GH_OST_TARGET_HOST=%s", he.migrationContext.GetTargetHostname()))
 	env = append(env, fmt.Sprintf("GH_OST_INSPECTED_LAG=%f", he.migrationContext.GetCurrentLagDuration().Seconds()))
 	env = append(env, fmt.Sprintf("GH_OST_HEARTBEAT_LAG=%f", he.migrationContext.TimeSinceLastHeartbeatOnChangelog().Seconds()))
 	env = append(env, fmt.Sprintf("GH_OST_PROGRESS=%f", he.migrationContext.GetProgressPct()))
@@ -242,6 +243,9 @@ func (he *HooksExecutor) applyEnvironmentVariables(extraVariables ...string) []s
 	env = append(env, fmt.Sprintf("GH_OST_HOOKS_HINT_TOKEN=%s", he.migrationContext.HooksHintToken))
 	env = append(env, fmt.Sprintf("GH_OST_DRY_RUN=%t", he.migrationContext.Noop))
 	env = append(env, fmt.Sprintf("GH_OST_REVERT=%t", he.migrationContext.Revert))
+	env = append(env, fmt.Sprintf("GH_OST_MOVE_TABLES=%t", he.migrationContext.IsMoveTablesMode()))
+	env = append(env, fmt.Sprintf("GH_OST_TARGET_DATABASE_NAME=%s", he.migrationContext.GetTargetDatabaseName()))
+	env = append(env, fmt.Sprintf("GH_OST_TARGET_TABLE_NAME=%s", he.migrationContext.GetTargetTableName()))
 
 	env = append(env, extraVariables...)
 	return env
@@ -320,8 +324,11 @@ func (he *HooksExecutor) OnInteractiveCommand(command string) error {
 }
 
 func (he *HooksExecutor) OnSuccess(instantDDL bool) error {
-	v := fmt.Sprintf("GH_OST_INSTANT_DDL=%t", instantDDL)
-	return he.executeHooks(onSuccess, v)
+	v := []string{fmt.Sprintf("GH_OST_INSTANT_DDL=%t", instantDDL)}
+	if he.migrationContext.IsMoveTablesMode() && he.migrationContext.MoveTables.DrainGTID != nil {
+		v = append(v, fmt.Sprintf("GH_OST_DRAIN_GTID=%s", he.migrationContext.MoveTables.DrainGTID.String()))
+	}
+	return he.executeHooks(onSuccess, v...)
 }
 
 func (he *HooksExecutor) OnFailure() error {

go/logic/hooks_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/github/gh-ost/go/base"
+	"github.com/github/gh-ost/go/mysql"
 )
 
 type recordingHooks struct {
@@ -242,9 +243,130 @@ func TestHooksExecutorExecuteHooks(t *testing.T) {
 				require.Equal(t, migrationContext.OriginalTableName, split[1])
 			case "GH_OST_INSTANT_DDL":
 				require.Equal(t, "false", split[1])
+			case "GH_OST_MOVE_TABLES":
+				require.Equal(t, "false", split[1])
+			case "GH_OST_TARGET_DATABASE_NAME":
+				require.Equal(t, migrationContext.DatabaseName, split[1])
+			case "GH_OST_TARGET_TABLE_NAME":
+				require.Equal(t, fmt.Sprintf("_%s_gho", migrationContext.OriginalTableName), split[1])
+			case "GH_OST_TARGET_HOST":
+				require.Equal(t, migrationContext.GetApplierHostname(), split[1])
 			case "TEST":
 				require.Equal(t, t.Name(), split[1])
 			}
 		}
 	})
 }
+
+func TestHooksExecutorMoveTablesEnvironmentVariables(t *testing.T) {
+	migrationContext := base.NewMigrationContext()
+	migrationContext.DatabaseName = "source_db"
+	migrationContext.OriginalTableName = "tablename"
+	migrationContext.MoveTables.TableNames = []string{"tablename"}
+	migrationContext.MoveTables.TargetDatabase = "target_db"
+	migrationContext.MoveTables.ConnectionConfig = mysql.NewConnectionConfig()
+	migrationContext.MoveTables.ConnectionConfig.Key.Hostname = "target.example.com"
+	migrationContext.MoveTables.ConnectionConfig.ImpliedKey = &migrationContext.MoveTables.ConnectionConfig.Key
+
+	hooksExecutor := NewHooksExecutor(migrationContext)
+
+	hooksPath, err := os.MkdirTemp("", "TestHooksExecutorMoveTablesEnvironmentVariables")
+	require.NoError(t, err)
+	defer os.RemoveAll(hooksPath)
+	migrationContext.HooksPath = hooksPath
+
+	require.NoError(t, os.WriteFile(
+		filepath.Join(hooksPath, "success-hook"),
+		[]byte("#!/bin/sh\nenv"),
+		0o777,
+	))
+
+	var buf bytes.Buffer
+	hooksExecutor.writer = &buf
+	require.Nil(t, hooksExecutor.executeHooks("success-hook"))
+
+	scanner := bufio.NewScanner(&buf)
+	for scanner.Scan() {
+		split := strings.SplitN(scanner.Text(), "=", 2)
+		switch split[0] {
+		case "GH_OST_MOVE_TABLES":
+			require.Equal(t, "true", split[1])
+		case "GH_OST_TARGET_DATABASE_NAME":
+			require.Equal(t, "target_db", split[1])
+		case "GH_OST_TARGET_TABLE_NAME":
+			require.Equal(t, "tablename", split[1])
+		case "GH_OST_TARGET_HOST":
+			require.Equal(t, "target.example.com", split[1])
+		}
+	}
+}
+
+func TestHooksExecutorOnSuccessEnvironmentVariables(t *testing.T) {
+	writeOnSuccessHook := func(t *testing.T, hooksPath string) {
+		t.Helper()
+		require.NoError(t, os.WriteFile(
+			filepath.Join(hooksPath, onSuccess),
+			[]byte("#!/bin/sh\nenv"),
+			0o777,
+		))
+	}
+
+	envFromBuffer := func(buf *bytes.Buffer) map[string]string {
+		envMap := map[string]string{}
+		scanner := bufio.NewScanner(buf)
+		for scanner.Scan() {
+			split := strings.SplitN(scanner.Text(), "=", 2)
+			if len(split) == 2 {
+				envMap[split[0]] = split[1]
+			}
+		}
+		return envMap
+	}
+
+	t.Run("move-tables-includes-drain-gtid", func(t *testing.T) {
+		migrationContext := base.NewMigrationContext()
+		migrationContext.DatabaseName = "source_db"
+		migrationContext.OriginalTableName = "tablename"
+		migrationContext.MoveTables.TableNames = []string{"tablename"}
+		migrationContext.MoveTables.DrainGTID = &mysql.FileBinlogCoordinates{LogFile: "mysql-bin.000001", LogPos: 12345}
+
+		hooksExecutor := NewHooksExecutor(migrationContext)
+
+		hooksPath, err := os.MkdirTemp("", "TestHooksExecutorOnSuccessEnvironmentVariables-move-tables")
+		require.NoError(t, err)
+		defer os.RemoveAll(hooksPath)
+		migrationContext.HooksPath = hooksPath
+		writeOnSuccessHook(t, hooksPath)
+
+		var buf bytes.Buffer
+		hooksExecutor.writer = &buf
+		require.NoError(t, hooksExecutor.OnSuccess(true))
+
+		envMap := envFromBuffer(&buf)
+		require.Equal(t, "true", envMap["GH_OST_INSTANT_DDL"])
+		require.Equal(t, migrationContext.MoveTables.DrainGTID.String(), envMap["GH_OST_DRAIN_GTID"])
+	})
+
+	t.Run("non-move-tables-omits-drain-gtid", func(t *testing.T) {
+		migrationContext := base.NewMigrationContext()
+		migrationContext.DatabaseName = "source_db"
+		migrationContext.OriginalTableName = "tablename"
+
+		hooksExecutor := NewHooksExecutor(migrationContext)
+
+		hooksPath, err := os.MkdirTemp("", "TestHooksExecutorOnSuccessEnvironmentVariables-standard")
+		require.NoError(t, err)
+		defer os.RemoveAll(hooksPath)
+		migrationContext.HooksPath = hooksPath
+		writeOnSuccessHook(t, hooksPath)
+
+		var buf bytes.Buffer
+		hooksExecutor.writer = &buf
+		require.NoError(t, hooksExecutor.OnSuccess(false))
+
+		envMap := envFromBuffer(&buf)
+		require.Equal(t, "false", envMap["GH_OST_INSTANT_DDL"])
+		_, exists := envMap["GH_OST_DRAIN_GTID"]
+		require.False(t, exists)
+	})
+}

go/logic/migrator.go

@@ -1082,6 +1082,7 @@ func (mgtr *Migrator) moveTablesCutOver() (err error) {
 	// OnSuccess call that used to live in MoveTables() (after finalCleanup) has
 	// been removed so the hook fires in the order coop_cutover.md §3.2 step 6
 	// requires (T5 between T4 and T6, BEFORE finalCleanup).
+	mgtr.migrationContext.MoveTables.DrainGTID = drainGTID
 	if err := mgtr.hooksExecutor.OnSuccess(false); err != nil {
 		return fmt.Errorf("on-success hook failed: %w", err)
 	}