feat: gh-ost 完整策略引擎 — 13项配置+环境差异+云数据库适配+DDL智能分析

配置项(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
This commit is contained in:
logan committed 2026-05-27 00:22:33 +08:00
1 parent 5d59ca427e
commit b2b13661fd
3 files changed
+253 -35

No files matched your search

+16
View File
@@ -28,6 +28,8 @@ type GhOstTask struct {
Execute bool `json:"execute"` Execute bool `json:"execute"`
ExactRowCount bool `json:"exact_row_count"` ExactRowCount bool `json:"exact_row_count"`
PostponeCutOver bool `json:"postpone_cut_over"` 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 // GhOstResult holds the execution result
@@ -96,6 +98,13 @@ func RunGhOst(ctx context.Context, task GhOstTask, ghOstPath string) (*GhOstResu
if task.PostponeCutOver { if task.PostponeCutOver {
args = append(args, "--postpone-cut-over-flag-file=/tmp/gh-ost-cutover.flag") 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 // Dry-Run vs Execute
// gh-ost defaults to noop (dry-run), only --execute actually applies changes // 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 { if task.PostponeCutOver {
args = append(args, "--postpone-cut-over-flag-file=/tmp/gh-ost-cutover.flag") 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 // gh-ost defaults to noop, only --execute actually applies changes
if task.Execute { if task.Execute {
+235 -35
View File
@@ -4,6 +4,7 @@ import (
"fmt" "fmt"
"net/http" "net/http"
"strconv" "strconv"
"strings"
"time" "time"
"sql-platform/internal/ghost" "sql-platform/internal/ghost"
@@ -12,6 +13,8 @@ import (
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
) )
// ==================== Ghost Settings Helpers ====================
// getSetting reads a system_setting value by key // getSetting reads a system_setting value by key
func getSetting(key, fallback string) string { func getSetting(key, fallback string) string {
var s model.SystemSetting var s model.SystemSetting
@@ -23,7 +26,7 @@ func getSetting(key, fallback string) string {
// isGhostEnabled checks if gh-ost is enabled in system settings // isGhostEnabled checks if gh-ost is enabled in system settings
func isGhostEnabled() bool { func isGhostEnabled() bool {
return getSetting("ghost_enabled", "true") == "true" return getSetting("ghost_enabled", "false") == "true"
} }
// getGhostPath returns the gh-ost binary path from settings // getGhostPath returns the gh-ost binary path from settings
@@ -31,6 +34,180 @@ func getGhostPath() string {
return getSetting("ghost_path", "gh-ost") 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 // POST /api/v1/ghost/dry-run
func GhOstDryRun(c *gin.Context) { func GhOstDryRun(c *gin.Context) {
if !isGhostEnabled() { if !isGhostEnabled() {
@@ -66,40 +243,6 @@ func GhOstDryRun(c *gin.Context) {
c.JSON(http.StatusOK, result) 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 // POST /api/v1/ghost/execute
func GhOstExecute(c *gin.Context) { func GhOstExecute(c *gin.Context) {
if !isGhostEnabled() { if !isGhostEnabled() {
@@ -112,6 +255,7 @@ func GhOstExecute(c *gin.Context) {
Database string `json:"database"` Database string `json:"database"`
Table string `json:"table"` Table string `json:"table"`
Alter string `json:"alter"` Alter string `json:"alter"`
Force bool `json:"force"` // skip auto-trigger check
} }
if err := c.ShouldBindJSON(&req); err != nil { if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) 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 // GET /api/v1/ghost/status/:taskId
func GhOstStatus(c *gin.Context) { func GhOstStatus(c *gin.Context) {
taskID := c.Param("taskId") taskID := c.Param("taskId")
+2
View File
@@ -141,6 +141,8 @@ func Setup(r *gin.Engine) {
ghostGroup := auth.Group("") ghostGroup := auth.Group("")
ghostGroup.Use(middleware.RequireRole("admin", "dba")) 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/dry-run", handler.GhOstDryRun)
ghostGroup.POST("/ghost/execute", handler.GhOstExecute) ghostGroup.POST("/ghost/execute", handler.GhOstExecute)
ghostGroup.GET("/ghost/status/:taskId", handler.GhOstStatus) ghostGroup.GET("/ghost/status/:taskId", handler.GhOstStatus)