feat(mysql): mysql清档重命名增加flush tables #8718

TencentBlueKing · Dec 23, 2024 · c2d0275 · c2d0275
1 parent 90d6b2b
commit c2d0275
Show file tree

Hide file tree

Showing 4 changed files with 52 additions and 6 deletions.
diff --git a/dbm-services/mysql/db-tools/dbactuator/pkg/components/rename_dbs/on_mysql.go b/dbm-services/mysql/db-tools/dbactuator/pkg/components/rename_dbs/on_mysql.go
@@ -4,15 +4,14 @@ import (
 	"bufio"
 	"bytes"
 	"context"
-	"fmt"
-	"os"
-	"os/exec"
-	"strings"
-
 	"dbm-services/common/go-pubpkg/logger"
 	"dbm-services/mysql/db-tools/dbactuator/pkg/components"
 	"dbm-services/mysql/db-tools/dbactuator/pkg/components/mysql/common"
 	"dbm-services/mysql/db-tools/dbactuator/pkg/components/rename_dbs/pkg"
+	"fmt"
+	"os"
+	"os/exec"
+	"strings"
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/jmoiron/sqlx"

diff --git a/dbm-services/mysql/db-tools/dbactuator/pkg/components/rename_dbs/pkg/drop_db.go b/dbm-services/mysql/db-tools/dbactuator/pkg/components/rename_dbs/pkg/drop_db.go
@@ -124,6 +124,12 @@ func isTableTransClean(conn *sqlx.Conn, from, to string) (bool, error) {
 		return false, err
 	}
 
+	defer func() {
+		for _, t := range tables {
+			flushTable(conn, from, to, t)
+		}
+	}()
+
 	for _, table := range tables {
 		yes, err := IsTableExistsIn(conn, table, to)
 		if err != nil {

diff --git a/dbm-services/mysql/db-tools/dbactuator/pkg/components/rename_dbs/pkg/trans_db_tables.go b/dbm-services/mysql/db-tools/dbactuator/pkg/components/rename_dbs/pkg/trans_db_tables.go
@@ -37,7 +37,32 @@ func TransDBTables(conn *sqlx.Conn, from, to string, tables []string) ([]string,
 	return res, nil
 }
 
+func flushTable(conn *sqlx.Conn, from, to, tableName string) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	_, err := conn.ExecContext(
+		ctx,
+		fmt.Sprintf("FLUSH TABLE `%s`.`%s`", from, tableName),
+	)
+	if err != nil {
+		logger.Error("flushing table `%s`.`%s` failed: %s", from, tableName, err.Error())
+	}
+
+	_, err = conn.ExecContext(
+		ctx,
+		fmt.Sprintf("FLUSH TABLE `%s`.`%s`", to, tableName),
+	)
+	if err != nil {
+		logger.Error("flushing table `%s`.`%s` failed: %s", to, tableName, err.Error())
+	}
+}
+
 func transDBTable(conn *sqlx.Conn, from, to, tableName string) ([]string, error) {
+	defer func() {
+		flushTable(conn, from, to, tableName)
+	}()
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 

diff --git a/dbm-services/mysql/db-tools/dbactuator/pkg/components/truncate/on_mysql.go b/dbm-services/mysql/db-tools/dbactuator/pkg/components/truncate/on_mysql.go
@@ -317,7 +317,6 @@ func (c *OnMySQLComponent) instanceRecreateSourceTables(port int) error {
 		}
 
 		// 这里改成从 stage 库拿表列表, 在重试的时候才能幂等
-		//for _, table := range c.dbTablesMap[db] {
 		for _, table := range stageTables {
 			err := c.instanceRecreateSourceTable(port, db, stageDBName, table)
 			if err != nil {
@@ -349,6 +348,10 @@ func (c *OnMySQLComponent) instanceRecreateSourceTable(port int, dbName, stageDB
 	//	return err
 	//}
 	//logger.Info("source table found in stage db on instance %d, try to truncate", port)
+	defer func() {
+		c.flushTable(dbName, tableName)
+		c.flushTable(stageDBName, tableName)
+	}()
 
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
@@ -370,6 +373,19 @@ func (c *OnMySQLComponent) instanceRecreateSourceTable(port int, dbName, stageDB
 	return nil
 }
 
+func (c *OnMySQLComponent) flushTable(dbName, tableName string) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	_, err := c.dbConn.ExecContext(
+		ctx,
+		fmt.Sprintf("FLUSH TABLE `%s`.`%s`", dbName, tableName),
+	)
+	if err != nil {
+		logger.Error("flush table %s.%s  failed: %s", dbName, tableName, err.Error())
+	}
+}
+
 func (c *OnMySQLComponent) GenerateDropStageSQL() error {
 	var dropSQLs []string
 	for dbName := range c.dbTablesMap {