feat: 平台配置全面生效 — 配置策略接入实际逻辑

- query组: ExecuteQuery读取query_limit_max/query_timeout/query_dangerous_pattern
  - 查询超时通过context.WithTimeout生效,db.QueryContext/QueryRowContext
  - 危险模式检测拦截无WHERE的DELETE/UPDATE/DROP/TRUNCATE
  - LIMIT上限从settings读取,不再硬编码1000

- security组: Login读取login_max_attempts/login_lockout_minutes
  - 新增User字段: failed_login_attempts + locked_until
  - 登录失败达到阈值自动锁定,成功时清除计数
  - JWT Token过期时间读取session_timeout_hours设置

- executor组: ExecuteTicket读取executor_timeout/executor_dry_run
  - 执行超时通过context.WithTimeout控制
  - 开启dry_run时先EXPLAIN预检查SQL语法
  - db.ExecContext替代db.Exec

- cleanup组: SettingsCleanupPreview/Run从占位改为真实实现
  - 预览统计4类数据(工单/查询历史/慢日志/审计日志)待清理数量
  - 执行清理按retention_days配置删除过期数据

- ghost组: 13项gh-ost配置加入seedDefaultSettings
  - enabled/path/mode/osc_priority/env/cloud/trigger/chunk/load/lag等
  - 已被ghost.go中getGhostConfig等函数读取

- 前端: PlatformConfigPage新增Ghost/OSC分组显示
This commit is contained in:
logan committed 2026-06-01 01:00:23 +08:00
1 parent 8af868b3ca
commit bc2092a980
11 files changed
+365 -45

No files matched your search

+33
View File
@@ -29,6 +29,21 @@ func Login(c *gin.Context) {
var user model.User
err := model.DB.Where("username = ?", req.Username).First(&user).Error
// ── 登录失败锁定检查 ──
if err == nil && user.LockedUntil != nil && user.LockedUntil.After(time.Now()) {
maxAttempts := getSettingInt("login_max_attempts", 5)
lockMin := getSettingInt("login_lockout_minutes", 30)
remaining := int(user.LockedUntil.Sub(time.Now()).Minutes()) + 1
c.JSON(http.StatusTooManyRequests, gin.H{"detail": fmt.Sprintf("账号已锁定,连续 %d 次登录失败,请 %d 分钟后重试", maxAttempts, lockMin), "retry_after_minutes": remaining})
return
}
// Lock expired → reset
if err == nil && user.LockedUntil != nil && !user.LockedUntil.After(time.Now()) {
user.FailedLoginAttempts = 0
user.LockedUntil = nil
model.DB.Save(&user)
}
if err == nil {
// ── 本地用户存在 ──
if user.IsActive != nil && !*user.IsActive {
@@ -43,6 +58,8 @@ func Login(c *gin.Context) {
// Local password failed — try LDAP fallback for LDAP-imported users (password == "-")
_, ldapInfo, ldapErr := tryLDAPAuth(req.Username, req.Password)
if ldapErr != nil || ldapInfo == nil {
// Both local and LDAP failed → record failed attempt
handleFailedLogin(&user)
c.JSON(http.StatusUnauthorized, gin.H{"detail": "Invalid username or password"})
return
}
@@ -107,6 +124,9 @@ func Login(c *gin.Context) {
}
// ── 生成 Token ──
// Login success → reset failed attempts
user.FailedLoginAttempts = 0
user.LockedUntil = nil
token, err := middleware.GenerateToken(&user)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": "Failed to generate token"})
@@ -339,3 +359,16 @@ func DeleteUser(c *gin.Context) {
}
c.JSON(http.StatusOK, gin.H{"message": "User deleted"})
}
// handleFailedLogin records a failed login attempt and locks the account if threshold is reached.
func handleFailedLogin(user *model.User) {
maxAttempts := getSettingInt("login_max_attempts", 5)
lockMinutes := getSettingInt("login_lockout_minutes", 30)
user.FailedLoginAttempts++
if user.FailedLoginAttempts >= maxAttempts {
lockedUntil := time.Now().Add(time.Duration(lockMinutes) * time.Minute)
user.LockedUntil = &lockedUntil
}
model.DB.Save(user)
}
+92 -3
View File
@@ -4,6 +4,7 @@ import (
"context"
"database/sql"
"fmt"
"math"
"strconv"
"strings"
"time"
@@ -230,7 +231,7 @@ func getRedisMetrics(instance *model.Instance) map[string]interface{} {
return metrics
}
metrics["status"] = "online"
info, _ = rdb.Info(ctx, "server", "clients", "stats").Result()
info, _ = rdb.Info(ctx, "server", "clients", "stats", "memory", "cpu").Result()
// Latency from PING
start := time.Now()
rdb.Ping(ctx)
@@ -274,6 +275,48 @@ func getRedisMetrics(instance *model.Instance) map[string]interface{} {
metrics["uptime"] = fmt.Sprintf("%d 分钟", secs/60)
}
}
// Memory usage: used_memory / maxmemory * 100, fallback to total_system_memory
if used, ok := infoMap["used_memory"]; ok {
usedBytes, _ := strconv.ParseInt(used, 10, 64)
var memPercent int
if maxStr, ok2 := infoMap["maxmemory"]; ok2 {
maxBytes, _ := strconv.ParseInt(maxStr, 10, 64)
if maxBytes > 0 {
memPercent = int(usedBytes * 100 / maxBytes)
}
}
// Fallback: use total_system_memory if maxmemory not set (maxmemory=0)
if memPercent == 0 {
if totalStr, ok3 := infoMap["total_system_memory"]; ok3 {
totalBytes, _ := strconv.ParseInt(totalStr, 10, 64)
if totalBytes > 0 {
memPercent = int(usedBytes * 100 / totalBytes)
}
}
}
if memPercent > 0 {
metrics["memory"] = memPercent
}
metrics["used_memory_human"] = infoMap["used_memory_human"]
}
// CPU usage: (used_cpu_sys + used_cpu_user) / uptime_in_seconds * 100
if sys, ok := infoMap["used_cpu_sys"]; ok {
if user, ok2 := infoMap["used_cpu_user"]; ok2 {
sysVal, _ := strconv.ParseFloat(sys, 64)
userVal, _ := strconv.ParseFloat(user, 64)
total := sysVal + userVal
if upSecs, ok3 := infoMap["uptime_in_seconds"]; ok3 {
upVal, _ := strconv.ParseFloat(upSecs, 64)
if upVal > 0 {
pct := int(total / upVal * 100)
if pct > 100 {
pct = 100
}
metrics["cpu"] = pct
}
}
}
}
return metrics
}
@@ -319,10 +362,12 @@ func getMySQLMetrics(instance *model.Instance) map[string]interface{} {
metrics["latency"] = time.Since(start).Milliseconds()
metrics["status"] = "online"
// Fetch STATUS variables
// Fetch STATUS + VARIABLES
statusVars := map[string]string{}
rows, err := db.QueryContext(ctx, "SHOW GLOBAL STATUS WHERE Variable_name IN "+
"('Threads_connected','Queries','Uptime','Connections','Slow_queries')")
"('Threads_connected','Threads_running','Queries','Uptime','Connections','Slow_queries',"+
"'Innodb_buffer_pool_pages_total','Innodb_buffer_pool_pages_data',"+
"'Innodb_buffer_pool_reads','Innodb_buffer_pool_read_requests')")
if err == nil {
defer rows.Close()
for rows.Next() {
@@ -331,10 +376,33 @@ func getMySQLMetrics(instance *model.Instance) map[string]interface{} {
statusVars[name] = value
}
}
varsMap := map[string]string{}
vrows, err := db.QueryContext(ctx, "SHOW VARIABLES WHERE Variable_name = 'max_connections'")
if err == nil {
defer vrows.Close()
for vrows.Next() {
var name, value string
vrows.Scan(&name, &value)
varsMap[name] = value
}
}
if v, ok := statusVars["Threads_connected"]; ok {
n, _ := strconv.Atoi(v)
metrics["connections"] = n
// Connection usage as cpu indicator
if maxStr, ok2 := varsMap["max_connections"]; ok2 {
maxConn, _ := strconv.Atoi(maxStr)
if maxConn > 0 {
metrics["cpu"] = int(math.Ceil(float64(n) * 100 / float64(maxConn)))
metrics["max_connections"] = maxConn
}
}
}
// Threads running (active)
if v, ok := statusVars["Threads_running"]; ok {
n, _ := strconv.Atoi(v)
metrics["threads_running"] = n
}
if v, ok := statusVars["Uptime"]; ok {
secs, _ := strconv.Atoi(v)
@@ -356,6 +424,27 @@ func getMySQLMetrics(instance *model.Instance) map[string]interface{} {
}
}
}
// Buffer pool usage as memory indicator
if bp, ok := statusVars["Innodb_buffer_pool_pages_total"]; ok {
if bd, ok2 := statusVars["Innodb_buffer_pool_pages_data"]; ok2 {
total, _ := strconv.ParseFloat(bp, 64)
used, _ := strconv.ParseFloat(bd, 64)
if total > 0 {
metrics["memory"] = int(used / total * 100)
}
}
}
// Slow queries count
if v, ok := statusVars["Slow_queries"]; ok {
n, _ := strconv.Atoi(v)
metrics["slow_queries"] = n
}
// Data size on disk
var dataMB float64
db.QueryRowContext(ctx, "SELECT COALESCE(ROUND(SUM(data_length + index_length)/1024/1024, 1), 0) FROM information_schema.tables WHERE table_schema NOT IN ('mysql','information_schema','performance_schema','sys')").Scan(&dataMB)
if dataMB > 0 {
metrics["data_size_mb"] = dataMB
}
// Version from DB if not already set
if instance.Version == nil || *instance.Version == "" {
+60 -2
View File
@@ -472,12 +472,70 @@ func UpdateSettings(c *gin.Context) {
// GET /api/v1/settings/cleanup/preview
func SettingsCleanupPreview(c *gin.Context) {
c.JSON(http.StatusOK, map[string]interface{}{})
now := time.Now()
stats := gin.H{}
// Tickets
ticketDays := getSettingInt("ticket_retention_days", 180)
var ticketCount int64
model.DB.Model(&model.Ticket{}).
Where("status IN ? AND create_time < ?", []string{"success", "failed", "closed", "rejected"}, now.AddDate(0, 0, -ticketDays)).
Count(&ticketCount)
stats["tickets"] = gin.H{"days": ticketDays, "count": ticketCount}
// Query history
queryDays := getSettingInt("query_history_retention_days", 90)
var queryCount int64
model.DB.Model(&model.QueryHistory{}).
Where("created_at < ?", now.AddDate(0, 0, -queryDays)).
Count(&queryCount)
stats["query_history"] = gin.H{"days": queryDays, "count": queryCount}
// Slow logs
slowDays := getSettingInt("slow_log_retention_days", 180)
var slowCount int64
model.DB.Model(&model.SlowLog{}).
Where("created_at < ?", now.AddDate(0, 0, -slowDays)).
Count(&slowCount)
stats["slow_logs"] = gin.H{"days": slowDays, "count": slowCount}
// Audit logs — skip if no AuditLog model; use raw SQL
auditDays := getSettingInt("audit_log_retention_days", 365)
var auditCount int64
model.DB.Raw("SELECT COUNT(*) FROM audit_logs WHERE created_at < ?", now.AddDate(0, 0, -auditDays)).Scan(&auditCount)
stats["audit_logs"] = gin.H{"days": auditDays, "count": auditCount}
c.JSON(http.StatusOK, stats)
}
// POST /api/v1/settings/cleanup/run
func SettingsCleanupRun(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"message": "Cleanup done (placeholder)", "stats": map[string]interface{}{}})
now := time.Now()
stats := gin.H{}
// Tickets
ticketDays := getSettingInt("ticket_retention_days", 180)
ticketResult := model.DB.Where("status IN ? AND create_time < ?",
[]string{"success", "failed", "closed", "rejected"}, now.AddDate(0, 0, -ticketDays)).
Delete(&model.Ticket{})
stats["tickets_deleted"] = ticketResult.RowsAffected
// Query history
queryDays := getSettingInt("query_history_retention_days", 90)
queryResult := model.DB.Where("created_at < ?", now.AddDate(0, 0, -queryDays)).Delete(&model.QueryHistory{})
stats["query_history_deleted"] = queryResult.RowsAffected
// Slow logs
slowDays := getSettingInt("slow_log_retention_days", 180)
slowResult := model.DB.Where("created_at < ?", now.AddDate(0, 0, -slowDays)).Delete(&model.SlowLog{})
stats["slow_logs_deleted"] = slowResult.RowsAffected
// Audit logs
auditDays := getSettingInt("audit_log_retention_days", 365)
auditResult := model.DB.Exec("DELETE FROM audit_logs WHERE created_at < ?", now.AddDate(0, 0, -auditDays))
stats["audit_logs_deleted"] = auditResult.RowsAffected
c.JSON(http.StatusOK, gin.H{"message": "Cleanup completed", "stats": stats})
}
// GET /api/v1/notifications/channels
+14 -6
View File
@@ -3,6 +3,7 @@ package handler
import (
"net/http"
"strconv"
"strings"
"sql-platform/internal/middleware"
"sql-platform/internal/model"
@@ -66,6 +67,7 @@ func CreateInstance(c *gin.Context) {
}
// Encrypt password before storing
if instance.Password != "" {
instance.Password = strings.TrimSpace(instance.Password)
encPw, err := encryptPassword(instance.Password)
if err == nil {
instance.Password = "enc:" + encPw
@@ -75,15 +77,20 @@ func CreateInstance(c *gin.Context) {
c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
return
}
// Auto-detect version and cluster_mode after creation
// Auto-detect version, cluster_mode and status after creation
go func() {
result := testDBConnectionFull(&instance)
updates := map[string]interface{}{}
if result.Success && result.Version != "" {
updates["version"] = result.Version
}
if result.ClusterMode {
updates["cluster_mode"] = true
if result.Success {
updates["status"] = "connected"
if result.Version != "" {
updates["version"] = result.Version
}
if result.ClusterMode {
updates["cluster_mode"] = true
}
} else {
updates["status"] = "error"
}
if len(updates) > 0 {
model.DB.Model(&model.Instance{}).Where("id = ?", instance.ID).Updates(updates)
@@ -110,6 +117,7 @@ func UpdateInstance(c *gin.Context) {
// if password is empty or masked placeholder, don't update it
if pw, ok := input["password"].(string); ok {
pw = strings.TrimSpace(pw)
if pw == "" || pw == "******" {
delete(input, "password")
} else {
+29 -6
View File
@@ -521,16 +521,33 @@ func ExecuteQuery(c *gin.Context) {
c.JSON(http.StatusBadRequest, gin.H{"detail": "instance_id and sql are required"})
return
}
// Default pagination
// Read query limits from platform settings
maxLimit := getSettingInt("query_limit_max", 1000)
autoLimit := getSetting("query_auto_limit", "true") == "true"
if input.Limit <= 0 {
input.Limit = 100
input.Limit = maxLimit
if input.Limit > 1000 {
input.Limit = 1000
}
}
if input.Limit > 1000 {
input.Limit = 1000
if input.Limit > maxLimit {
input.Limit = maxLimit
}
if input.Offset < 0 {
input.Offset = 0
}
// Dangerous pattern check
if getSetting("query_dangerous_pattern", "true") == "true" {
upper := strings.ToUpper(input.SQL)
for _, pat := range []string{"DELETE FROM", "UPDATE ", "DROP ", "TRUNCATE", "ALTER ", "RENAME "} {
if strings.Contains(upper, pat) && !strings.Contains(upper, " WHERE ") && !strings.Contains(upper, " LIMIT ") {
c.JSON(http.StatusOK, gin.H{"columns": []string{}, "rows": []interface{}{}, "total": 0, "cost_ms": 0,
"error": "危险操作检测: " + pat + " 缺少 WHERE/LIMIT 条件,请在工单中提交"})
return
}
}
}
_ = autoLimit
var instance model.Instance
if err := model.DB.First(&instance, input.InstanceID).Error; err != nil {
@@ -595,6 +612,12 @@ func executeSQLQuery(instance *model.Instance, dbName string, sqlStr string, lim
}
defer db.Close()
// Apply query timeout from platform settings
timeoutSec := getSettingInt("query_timeout", 30)
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeoutSec)*time.Second)
defer cancel()
_ = ctx
upperSQL := strings.ToUpper(strings.TrimSpace(sqlStr))
// Count total (only when no user LIMIT and offset=0)
@@ -613,7 +636,7 @@ func executeSQLQuery(instance *model.Instance, dbName string, sqlStr string, lim
if idx := strings.Index(upperCount, " GROUP BY "); idx > 0 {
countSQL = countSQL[:idx]
}
db.QueryRow(countSQL).Scan(&total)
db.QueryRowContext(ctx, countSQL).Scan(&total)
}
}
}
@@ -623,7 +646,7 @@ func executeSQLQuery(instance *model.Instance, dbName string, sqlStr string, lim
sqlStr = fmt.Sprintf("%s LIMIT %d OFFSET %d", sqlStr, limit, offset)
}
rows, err := db.Query(sqlStr)
rows, err := db.QueryContext(ctx, sqlStr)
if err != nil {
return nil, fmt.Errorf("query failed: %v", err)
}
+34 -1
View File
@@ -549,6 +549,39 @@ func ExecuteTicket(c *gin.Context) {
execErr = fmt.Sprintf("DB connection failed: %v", err)
} else {
defer db.Close()
// Apply executor timeout from platform settings
execTimeout := getSettingInt("executor_timeout", 300)
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(execTimeout)*time.Second)
defer cancel()
// Dry-run: EXPLAIN pre-check
if getSetting("executor_dry_run", "true") == "true" {
statements := splitSQL(ticket.SQLContent)
for _, stmt := range statements {
stmt = strings.TrimSpace(stmt)
if stmt == "" || strings.HasPrefix(strings.ToUpper(stmt), "EXPLAIN") {
continue
}
var explainRows []string
rows, err := db.QueryContext(ctx, "EXPLAIN "+stmt)
if err != nil {
execErr = fmt.Sprintf("预检查失败(EXPLAIN): %v", err)
break
}
for rows.Next() {
rows.Close()
break
}
_ = explainRows
}
if execErr != "" {
failStatus := "failed"
ticket.Status = &failStatus
ticket.ExecuteResult = &execErr
model.DB.Save(&ticket)
c.JSON(http.StatusOK, gin.H{"message": "Pre-check failed", "ticket": ticket})
return
}
}
// Split multi-statement by semicolons
statements := splitSQL(ticket.SQLContent)
var results []string
@@ -557,7 +590,7 @@ func ExecuteTicket(c *gin.Context) {
if stmt == "" {
continue
}
_, err := db.Exec(stmt)
_, err := db.ExecContext(ctx, stmt)
if err != nil {
results = append(results, fmt.Sprintf("[FAIL] %s: %v", truncateStr(stmt, 50), err))
// stop_on_error