From b2b13661fd14eed9c76c5ca5b1baba99c66ddfa3 Mon Sep 17 00:00:00 2001 From: logan Date: Wed, 27 May 2026 00:22:33 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20gh-ost=20=E5=AE=8C=E6=95=B4=E7=AD=96?= =?UTF-8?q?=E7=95=A5=E5=BC=95=E6=93=8E=20=E2=80=94=2013=E9=A1=B9=E9=85=8D?= =?UTF-8?q?=E7=BD=AE+=E7=8E=AF=E5=A2=83=E5=B7=AE=E5=BC=82+=E4=BA=91?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=BA=93=E9=80=82=E9=85=8D+DDL=E6=99=BA?= =?UTF-8?q?=E8=83=BD=E5=88=86=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 配置项(13项): - ghost_enabled: 全局开关(默认关闭) - ghost_default_mode: 默认执行模式(dry_run/execute) - ghost_osc_priority: OSC优先级(native_first/ghost_first/ghost_only) - ghost_env_mode: 环境模式(development/production) - ghost_cloud_type: 云数据库类型(self_hosted/aliyun_rds/polardb/aws_aurora) - ghost_auto_trigger_rows: 自动触发行数阈值(万) - ghost_auto_trigger_ddl: 自动触发DDL类型 - ghost_path/chunk_size/max_load/critical_load/max_lag_millis/allow_on_master 新增API: - GET /ghost/config: 查看全部gh-ost配置 - POST /ghost/analyze: DDL智能分析(解析DDL类型+行数检查+策略推荐) - 云数据库: --aliyun-rds / --aws 自动适配 策略逻辑: - native_first + development → 推荐Native DDL - native_first + production + DDL匹配 → 推荐gh-ost - ghost_first/ghost_only → 强制gh-ost - 行数超阈值 → 额外推荐gh-ost --- backend/internal/ghost/ghost.go | 16 ++ backend/internal/handler/ghost.go | 270 ++++++++++++++++++++++++++---- backend/router/router.go | 2 + 3 files changed, 253 insertions(+), 35 deletions(-) diff --git a/backend/internal/ghost/ghost.go b/backend/internal/ghost/ghost.go index d4564b4..b80e1a6 100644 --- a/backend/internal/ghost/ghost.go +++ b/backend/internal/ghost/ghost.go @@ -28,6 +28,8 @@ type GhOstTask struct { Execute bool `json:"execute"` ExactRowCount bool `json:"exact_row_count"` PostponeCutOver bool `json:"postpone_cut_over"` + AliyunRDS bool `json:"aliyun_rds,omitempty"` // --aliyun-rds + AWSAurora bool `json:"aws_aurora,omitempty"` // --aws } // GhOstResult holds the execution result @@ -96,6 +98,13 @@ func RunGhOst(ctx context.Context, task GhOstTask, ghOstPath string) (*GhOstResu if task.PostponeCutOver { args = append(args, "--postpone-cut-over-flag-file=/tmp/gh-ost-cutover.flag") } + // Cloud-specific flags + if task.AliyunRDS { + args = append(args, "--aliyun-rds") + } + if task.AWSAurora { + args = append(args, "--aws") + } // Dry-Run vs Execute // gh-ost defaults to noop (dry-run), only --execute actually applies changes @@ -235,6 +244,13 @@ func buildArgs(task GhOstTask, ghOstPath string) []string { if task.PostponeCutOver { args = append(args, "--postpone-cut-over-flag-file=/tmp/gh-ost-cutover.flag") } + // Cloud-specific flags + if task.AliyunRDS { + args = append(args, "--aliyun-rds") + } + if task.AWSAurora { + args = append(args, "--aws") + } // gh-ost defaults to noop, only --execute actually applies changes if task.Execute { diff --git a/backend/internal/handler/ghost.go b/backend/internal/handler/ghost.go index 0fcfb97..7f49db5 100644 --- a/backend/internal/handler/ghost.go +++ b/backend/internal/handler/ghost.go @@ -4,6 +4,7 @@ import ( "fmt" "net/http" "strconv" + "strings" "time" "sql-platform/internal/ghost" @@ -12,6 +13,8 @@ import ( "github.com/gin-gonic/gin" ) +// ==================== Ghost Settings Helpers ==================== + // getSetting reads a system_setting value by key func getSetting(key, fallback string) string { var s model.SystemSetting @@ -23,7 +26,7 @@ func getSetting(key, fallback string) string { // isGhostEnabled checks if gh-ost is enabled in system settings func isGhostEnabled() bool { - return getSetting("ghost_enabled", "true") == "true" + return getSetting("ghost_enabled", "false") == "true" } // getGhostPath returns the gh-ost binary path from settings @@ -31,6 +34,180 @@ func getGhostPath() string { return getSetting("ghost_path", "gh-ost") } +// getGhostConfig returns all ghost-related settings as a map for API responses +func getGhostConfig() gin.H { + return gin.H{ + "enabled": isGhostEnabled(), + "default_mode": getSetting("ghost_default_mode", "dry_run"), + "osc_priority": getSetting("ghost_osc_priority", "native_first"), + "env_mode": getSetting("ghost_env_mode", "development"), + "cloud_type": getSetting("ghost_cloud_type", "self_hosted"), + "auto_trigger_rows": getSettingInt("ghost_auto_trigger_rows", 10), + "auto_trigger_ddl": getSetting("ghost_auto_trigger_ddl", "ADD COLUMN,MODIFY COLUMN,ADD INDEX"), + "chunk_size": getSettingInt("ghost_default_chunk_size", 1000), + "max_load": getSetting("ghost_max_load", "Threads_running=25"), + "critical_load": getSetting("ghost_critical_load", "Threads_running=50"), + "allow_on_master": getSetting("ghost_allow_on_master", "true") == "true", + "max_lag_millis": getSettingInt("ghost_max_lag_millis", 1500), + } +} + +// getSettingInt reads a system_setting as integer +func getSettingInt(key string, fallback int) int { + if v, err := strconv.Atoi(getSetting(key, "")); err == nil { + return v + } + return fallback +} + +// ==================== Ghost OSC Strategy Engine ==================== + +// DDLAnalysis contains the analysis result of an ALTER statement +type DDLAnalysis struct { + DDLTypes []string // extracted DDL operation types + NeedsOSC bool // whether OSC tool is recommended + Reason string // explanation for the recommendation + RecommendedTool string // "native", "gh-ost", "pt-osc" +} + +// analyzeDDL inspects the ALTER statement to determine DDL types and OSC need +func analyzeDDL(alterStmt string) DDLAnalysis { + upper := strings.ToUpper(alterStmt) + var types []string + + ddls := map[string]string{ + "ADD COLUMN": "ADD COLUMN", "DROP COLUMN": "DROP COLUMN", + "MODIFY COLUMN": "MODIFY COLUMN", "CHANGE COLUMN": "CHANGE COLUMN", + "ADD INDEX": "ADD INDEX", "DROP INDEX": "DROP INDEX", + "ADD UNIQUE": "ADD INDEX", "ADD PRIMARY KEY": "ADD INDEX", + "DROP PRIMARY KEY": "DROP INDEX", + "RENAME COLUMN": "CHANGE COLUMN", "ALGORITHM": "ALGORITHM", + } + for pattern, ddlType := range ddls { + if strings.Contains(upper, pattern) { + types = appendIfMissing(types, ddlType) + } + } + + // Determine recommendation based on config + oscPriority := getSetting("ghost_osc_priority", "native_first") + envMode := getSetting("ghost_env_mode", "development") + + analysis := DDLAnalysis{DDLTypes: types, RecommendedTool: "native"} + + switch oscPriority { + case "ghost_only": + analysis.NeedsOSC = true + analysis.RecommendedTool = "gh-ost" + analysis.Reason = "平台配置为仅使用 gh-ost 模式" + case "ghost_first": + analysis.NeedsOSC = true + analysis.RecommendedTool = "gh-ost" + analysis.Reason = "平台配置为优先使用 gh-ost" + default: // native_first + if envMode == "production" { + // Production: check auto-trigger conditions + triggerDDLs := strings.Split(getSetting("ghost_auto_trigger_ddl", ""), ",") + for _, t := range types { + for _, trigger := range triggerDDLs { + if strings.TrimSpace(trigger) == t { + analysis.NeedsOSC = true + analysis.RecommendedTool = "gh-ost" + analysis.Reason = fmt.Sprintf("生产环境 + DDL类型[%s]匹配自动触发规则", t) + break + } + } + } + } else { + analysis.Reason = "开发/测试环境，建议使用 Native DDL" + } + } + + return analysis +} + +// appendIfMissing adds a string to slice only if not present +func appendIfMissing(slice []string, s string) []string { + for _, el := range slice { + if el == s { + return slice + } + } + return append(slice, s) +} + +// ==================== Ghost API Handlers ==================== + +// GET /api/v1/ghost/config — return current ghost configuration +func GhOstGetConfig(c *gin.Context) { + c.JSON(http.StatusOK, getGhostConfig()) +} + +// POST /api/v1/ghost/analyze — analyze DDL and return OSC recommendation +func GhOstAnalyzeDDL(c *gin.Context) { + var req struct { + InstanceID int `json:"instance_id"` + Database string `json:"database"` + Table string `json:"table"` + Alter string `json:"alter"` + } + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) + return + } + + analysis := analyzeDDL(req.Alter) + + // Also check table row count if instance_id provided + rowCount := -1 + if req.InstanceID > 0 { + rowCount = getTableRowCount(req.InstanceID, req.Database, req.Table) + } + + // Check auto-trigger by row count + triggerRows := getSettingInt("ghost_auto_trigger_rows", 10) * 10000 + if rowCount >= 0 && rowCount > triggerRows && !analysis.NeedsOSC { + analysis.NeedsOSC = true + analysis.RecommendedTool = "gh-ost" + analysis.Reason = fmt.Sprintf("表行数(%d)超过阈值(%d)，推荐使用 gh-ost", rowCount, triggerRows) + } + + c.JSON(http.StatusOK, gin.H{ + "enabled": isGhostEnabled(), + "ddl_types": analysis.DDLTypes, + "needs_osc": analysis.NeedsOSC, + "recommended_tool": analysis.RecommendedTool, + "reason": analysis.Reason, + "table_rows": rowCount, + "trigger_rows": triggerRows, + "config": getGhostConfig(), + }) +} + +// getTableRowCount estimates row count via INFORMATION_SCHEMA +func getTableRowCount(instanceID int, database, table string) int { + var inst model.Instance + if model.DB.First(&inst, instanceID).Error != nil { + return -1 + } + if inst.DBType != "mysql" { + return -1 + } + + db, err := getDBConnection(&inst, database) + if err != nil { + return -1 + } + defer db.Close() + + var count int + row := db.QueryRow("SELECT TABLE_ROWS FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ?", database, table) + if err := row.Scan(&count); err != nil { + return -1 + } + return count +} + // POST /api/v1/ghost/dry-run func GhOstDryRun(c *gin.Context) { if !isGhostEnabled() { @@ -66,40 +243,6 @@ func GhOstDryRun(c *gin.Context) { c.JSON(http.StatusOK, result) } -// buildGhOstTask resolves instance_id to connection params -func buildGhOstTask(instanceID int, database, table, alter string) (*ghost.GhOstTask, error) { - var inst model.Instance - if err := model.DB.First(&inst, instanceID).Error; err != nil { - return nil, fmt.Errorf("instance %d not found", instanceID) - } - if inst.DBType != "mysql" { - return nil, fmt.Errorf("gh-ost only supports MySQL instances, got %s", inst.DBType) - } - return &ghost.GhOstTask{ - Host: inst.Host, - Port: inst.Port, - User: inst.Username, - Password: inst.Password, - Database: database, - Table: table, - Alter: alter, - }, nil -} - -// applyGhostSettings applies system_settings to the task (chunk-size, max-load, cloud flags) -func applyGhostSettings(task *ghost.GhOstTask) { - // Chunk size - if cs, err := strconv.Atoi(getSetting("ghost_default_chunk_size", "1000")); err == nil && cs > 0 { - task.ChunkSize = cs - } - // Max load - if ml := getSetting("ghost_max_load", ""); ml != "" { - task.MaxLoad = ml - } - // Allow on master - task.AllowOnMaster = getSetting("ghost_allow_on_master", "true") == "true" -} - // POST /api/v1/ghost/execute func GhOstExecute(c *gin.Context) { if !isGhostEnabled() { @@ -112,6 +255,7 @@ func GhOstExecute(c *gin.Context) { Database string `json:"database"` Table string `json:"table"` Alter string `json:"alter"` + Force bool `json:"force"` // skip auto-trigger check } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) @@ -134,6 +278,62 @@ func GhOstExecute(c *gin.Context) { }) } +// ==================== Ghost Helpers ==================== + +// buildGhOstTask resolves instance_id to connection params +func buildGhOstTask(instanceID int, database, table, alter string) (*ghost.GhOstTask, error) { + var inst model.Instance + if err := model.DB.First(&inst, instanceID).Error; err != nil { + return nil, fmt.Errorf("instance %d not found", instanceID) + } + if inst.DBType != "mysql" { + return nil, fmt.Errorf("gh-ost only supports MySQL instances, got %s", inst.DBType) + } + + task := &ghost.GhOstTask{ + Host: inst.Host, + Port: inst.Port, + User: inst.Username, + Password: inst.Password, + Database: database, + Table: table, + Alter: alter, + } + + // Apply cloud-specific flags + cloudType := getSetting("ghost_cloud_type", "self_hosted") + switch cloudType { + case "aliyun_rds": + task.AliyunRDS = true + case "aws_aurora": + task.AWSAurora = true + } + + return task, nil +} + +// applyGhostSettings applies all system_settings to the task +func applyGhostSettings(task *ghost.GhOstTask) { + // Chunk size + if cs := getSettingInt("ghost_default_chunk_size", 1000); cs > 0 { + task.ChunkSize = cs + } + // Max load + if ml := getSetting("ghost_max_load", ""); ml != "" { + task.MaxLoad = ml + } + // Critical load + if cl := getSetting("ghost_critical_load", ""); cl != "" { + task.CriticalLoad = cl + } + // Max lag millis + if ml := getSettingInt("ghost_max_lag_millis", 1500); ml > 0 { + task.MaxLagMillis = ml + } + // Allow on master + task.AllowOnMaster = getSetting("ghost_allow_on_master", "true") == "true" +} + // GET /api/v1/ghost/status/:taskId func GhOstStatus(c *gin.Context) { taskID := c.Param("taskId") diff --git a/backend/router/router.go b/backend/router/router.go index 004b9dc..561d971 100644 --- a/backend/router/router.go +++ b/backend/router/router.go @@ -141,6 +141,8 @@ func Setup(r *gin.Engine) { ghostGroup := auth.Group("") ghostGroup.Use(middleware.RequireRole("admin", "dba")) { + ghostGroup.GET("/ghost/config", handler.GhOstGetConfig) + ghostGroup.POST("/ghost/analyze", handler.GhOstAnalyzeDDL) ghostGroup.POST("/ghost/dry-run", handler.GhOstDryRun) ghostGroup.POST("/ghost/execute", handler.GhOstExecute) ghostGroup.GET("/ghost/status/:taskId", handler.GhOstStatus)