Files
opsxsql/backend/internal/handler/ghost.go
T
opsx 0c57e18f19 feat(log): Gin 改结构化 JSON 访问日志 + 保留期自动清理(含 VACUUM 回收)
=== 一、审计结论(用户问:gin 日志是否标准、会不会打爆磁盘)===
现状是标准的:gin.Default() 写 stdout,代码里没有任何
gin.DefaultWriter / log.SetOutput / lumberjack / os.OpenFile 写日志文件;
容器内 /var/log 是空的,find / -name "*.log" 无结果;
kubelet containerLogMaxSize=10Mi / containerLogMaxFiles=5 自动轮转,
实测 /var/log 只占 120G 磁盘的 0.8%。

但发现两个真问题:
  1. 访问日志是 gin 默认文本格式,Loki 里没法按字段过滤
  2. 审计日志/查询历史等只配了保留天数,**没有任何自动清理** ——
     只有手动按钮,表会一直涨,SQLite 文件删了也不缩

=== 二、结构化 JSON 访问日志(middleware/accesslog.go)===
gin.Default() → gin.New() + AccessLog() + gin.Recovery() + CORSMiddleware
  - 一行一个 JSON:ts/level/type/method/path/status/latency_ms/bytes/ip/ua
  - 自动分级:5xx=error / 4xx=warn / 其余=info
  - 鉴权后带 user_id / username,便于审计
  - 健康检查与 /assets 静音(K8s 探针每 10s 一次,会淹没真实请求)
  - Recovery 显式加回(gin.Default 里包含它,漏了 panic 会打挂进程)

⚠️ 关键细节:Go 的 map 序列化按 key 字母序,ts 会排到最后,
   一行以 {"bytes":... 开头根本看不出时间。LogJSON 手工拼装,
   把 ts/level/type/msg 固定在最前,其余排序输出。

=== 三、保留期自动清理(handler/cleanup.go)===
启动 2 分钟后跑一次,之后每 24h 一次;cleanup_auto_enabled=false 可关。

覆盖 8 张会无限增长的表(原来只管 4 张):
  audit_logs / query_histories / slow_logs / tickets
  + operation_logs / workflow_execution_logs / scheduled_task_logs(新增)
  + refresh_tokens(按 expires_at < now,过期不清理会每次登录加一行)

设计取舍:
  - 每批删 5000 行、批间 sleep 50ms:SQLite 单写者,一条 DELETE 全删会把库锁死
  - 单表单轮最多 20 万行,剩下的留给下一轮
  - deleted 后看 freelist/page_count,>25% 才 VACUUM(DELETE 不还磁盘,必须 VACUUM)
  - 保留天数 <=0 视为不清理,避免误配把数据全删

手动接口同步改造:/settings/cleanup/{preview,run} 复用同一套逻辑,
返回结构改为 {deleted, total, vacuumed, freed_bytes, cost_ms}。

新增系统设置(seed 自动补,已实测 8 个 key 全部落库):
  operation_log_retention_days=365  workflow_log_retention_days=180
  scheduled_log_retention_days=180  cleanup_auto_enabled=true

=== 四、实测 ===
访问日志:{"ts":"2026-10-27T14:46:39.58Z","level":"info","type":"access","bytes":668,"ip":"...","latency_ms":78.6,"method":"POST","path":"/api/v1/auth/login","status":200,...}
health 日志行数: 0 ✅   [GIN] 文本行数: 0 ✅   非 JSON 行: 4(仅启动)
cleanup preview: 8 张表的 count/days ✅
cleanup run: {"cost_ms":1,"deleted":{},"freed_bytes":0,"vacuumed":false,"total":0} ✅

镜像:lghost/opsxsql:latest + v0.1 = sha256:30593de3fc09
2026-10-27 14:49:45 +00:00

737 lines
23 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package handler
import (
"fmt"
"net/http"
"strconv"
"strings"
"time"
"sql-platform/internal/ghost"
"sql-platform/internal/handler/resp"
"sql-platform/internal/middleware"
"sql-platform/internal/model"
"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
if model.DB.Where("`key` = ?", key).Take(&s).Error == nil && s.Value != "" {
return s.Value
}
return fallback
}
// isGhostEnabled checks if gh-ost is enabled in system settings
func isGhostEnabled() bool {
return getSetting("ghost_enabled", "false") == "true"
}
// getGhostPath returns the gh-ost binary path from settings
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())
}
// getMySQLVersion returns the MySQL version string for an instance
func getMySQLVersion(instanceID int) string {
var inst model.Instance
if model.DB.First(&inst, instanceID).Error != nil {
return ""
}
if inst.DBType != "mysql" {
return ""
}
db, err := getDBConnection(&inst, "")
if err != nil {
return ""
}
defer db.Close()
var version string
db.QueryRow("SELECT VERSION()").Scan(&version)
return version
}
// supportsInstantDDL checks if MySQL version supports INSTANT DDL (8.0.12+)
func supportsInstantDDL(version string) bool {
// Parse "8.0.32" or "5.7.40-log" etc
parts := strings.SplitN(version, "-", 2)
v := parts[0]
segments := strings.Split(v, ".")
if len(segments) < 3 {
return false
}
major, _ := strconv.Atoi(segments[0])
minor, _ := strconv.Atoi(segments[1])
patch, _ := strconv.Atoi(segments[2])
if major > 8 {
return true
}
if major == 8 && minor > 0 {
return true
}
if major == 8 && minor == 0 && patch >= 12 {
return true
}
return false
}
// 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 {
resp.BadRequest(c, err.Error())
return
}
analysis := analyzeDDL(req.Alter)
// Get MySQL version for INSTANT DDL check
mysqlVersion := ""
instantDDL := false
if req.InstanceID > 0 {
mysqlVersion = getMySQLVersion(req.InstanceID)
instantDDL = supportsInstantDDL(mysqlVersion)
}
// Cloud databases: prefer Native DDL
cloudType := getSetting("ghost_cloud_type", "self_hosted")
if cloudType != "self_hosted" && analysis.NeedsOSC {
analysis.NeedsOSC = false
analysis.RecommendedTool = "native"
analysis.Reason = fmt.Sprintf("云数据库(%s)建议使用原生DDL,无需gh-ost", cloudType)
}
// 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 (only for self-hosted)
if cloudType == "self_hosted" {
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,
"mysql_version": mysqlVersion,
"supports_instant": instantDDL,
"cloud_type": cloudType,
"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() {
resp.Forbidden(c, "gh-ost 功能已禁用,请在系统设置中开启")
return
}
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 {
resp.BadRequest(c, err.Error())
return
}
task, err := buildGhOstTask(req.InstanceID, req.Database, req.Table, req.Alter)
if err != nil {
resp.BadRequest(c, err.Error())
return
}
task.DryRun = true
task.Execute = false
applyGhostSettings(task)
result, err := ghost.RunGhOst(c.Request.Context(), *task, getGhostPath())
if err != nil {
c.JSON(http.StatusInternalServerError, result)
return
}
c.JSON(http.StatusOK, result)
}
// POST /api/v1/ghost/execute
func GhOstExecute(c *gin.Context) {
if !isGhostEnabled() {
resp.Forbidden(c, "gh-ost 功能已禁用,请在系统设置中开启")
return
}
var req struct {
InstanceID int `json:"instance_id"`
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 {
resp.BadRequest(c, err.Error())
return
}
task, err := buildGhOstTask(req.InstanceID, req.Database, req.Table, req.Alter)
if err != nil {
resp.BadRequest(c, err.Error())
return
}
task.DryRun = false
task.Execute = true
applyGhostSettings(task)
taskID := ghost.RunGhOstAsync(*task, getGhostPath(), nil)
c.JSON(http.StatusAccepted, gin.H{
"task_id": taskID,
"message": "gh-ost execution started",
})
}
// ==================== 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)
}
// Decrypt password for gh-ost command line
password := inst.Password
if dec, err := decryptPassword(strings.TrimPrefix(inst.Password, "enc:")); err == nil {
password = dec
}
task := &ghost.GhOstTask{
Host: inst.Host,
Port: inst.Port,
User: inst.Username,
Password: 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")
status, result := ghost.Manager.GetTaskStatus(taskID)
if status == "" {
resp.NotFound(c, "Task not found")
return
}
c.JSON(http.StatusOK, gin.H{
"task_id": taskID,
"status": status,
"result": result,
})
}
// GET /api/v1/ghost/tasks
func GhOstListTasks(c *gin.Context) {
tasks := ghost.Manager.ListTasks()
c.JSON(http.StatusOK, tasks)
}
// GET /api/v1/audit-rules
func ListAuditRules(c *gin.Context) {
var rules []model.AuditRule
query := model.DB.Order("id")
if dbType := c.Query("db_type"); dbType != "" {
query = query.Where("db_type = ?", dbType)
}
query.Find(&rules)
c.JSON(http.StatusOK, rules)
}
// GET /api/v1/announcements
func ListAnnouncements(c *gin.Context) {
var announcements []model.Announcement
model.DB.Where("is_published = ?", true).Order("id DESC").Find(&announcements)
c.JSON(http.StatusOK, announcements)
}
// GET /api/v1/query-histories
func ListQueryHistories(c *gin.Context) {
userID, _, _ := middleware.GetCurrentUser(c)
q := model.DB.Order("id DESC").Limit(100)
if c.Query("mine") == "1" {
q = q.Where("user_id = ?", userID)
}
var histories []model.QueryHistory
q.Find(&histories)
resp.OK(c, histories)
}
// PUT /api/v1/query-histories/:id/favorite — 收藏/取消收藏
func ToggleQueryHistoryFavorite(c *gin.Context) {
userID, _, _ := middleware.GetCurrentUser(c)
id := c.Param("id")
var history model.QueryHistory
if err := model.DB.First(&history, id).Error; err != nil {
resp.NotFound(c, "History not found")
return
}
if history.UserID != userID {
resp.Forbidden(c, "只能收藏自己的查询历史")
return
}
if err := model.DB.Model(&history).Update("is_favorite", !history.IsFavorite).Error; err != nil {
resp.InternalErr(c, err)
return
}
resp.OK(c, gin.H{"id": history.ID, "is_favorite": !history.IsFavorite})
}
// GET /api/v1/settings — returns settings grouped by `group` field
func ListSettings(c *gin.Context) {
var settings []model.SystemSetting
model.DB.Order("sort_order").Find(&settings)
// group by `group` field
grouped := make(map[string][]model.SystemSetting)
for _, s := range settings {
grouped[s.Group] = append(grouped[s.Group], s)
}
c.JSON(http.StatusOK, grouped)
}
// PUT /api/v1/settings — batch update settings
func UpdateSettings(c *gin.Context) {
var items []struct {
Key string `json:"key"`
Value string `json:"value"`
}
if err := c.ShouldBindJSON(&items); err != nil {
resp.BadRequest(c, err.Error())
return
}
for _, item := range items {
model.DB.Model(&model.SystemSetting{}).Where("`key` = ?", item.Key).Update("value", item.Value)
}
c.JSON(http.StatusOK, gin.H{"message": "Settings updated"})
}
// GET /api/v1/settings/cleanup/preview
//
// 统计口径统一收敛到 CleanupPreview()(handler/cleanup.go),
// 与后台定时清理共用同一份规则表,避免「预览说有、真跑不删」这类漂移。
func SettingsCleanupPreview(c *gin.Context) {
c.JSON(http.StatusOK, CleanupPreview())
}
// POST /api/v1/settings/cleanup/run
//
// 手动触发:与后台定时任务走同一套分批删除 + VACUUM 逻辑。
func SettingsCleanupRun(c *gin.Context) {
start := time.Now()
deleted := RunRetentionCleanup()
vacuumed, freed := VacuumIfNeeded()
var total int64
for _, n := range deleted {
total += n
}
c.JSON(http.StatusOK, gin.H{
"message": "Cleanup completed",
"deleted": deleted,
"total": total,
"vacuumed": vacuumed,
"freed_bytes": freed,
"cost_ms": time.Since(start).Milliseconds(),
})
}
// GET /api/v1/notifications/channels
func ListNotificationChannels(c *gin.Context) {
var channels []model.NotificationChannel
model.DB.Order("id").Find(&channels)
c.JSON(http.StatusOK, channels)
}
// GET /api/v1/notifications/templates
func ListNotificationTemplates(c *gin.Context) {
var templates []model.NotificationTemplate
model.DB.Order("id").Find(&templates)
c.JSON(http.StatusOK, templates)
}
// GET /api/v1/workflows
func ListWorkflows(c *gin.Context) {
var workflows []model.WorkflowDefinition
model.DB.Order("id").Find(&workflows)
type WorkflowWithSteps struct {
model.WorkflowDefinition
Steps []model.WorkflowStep `json:"steps"`
}
result := make([]WorkflowWithSteps, len(workflows))
for i, wf := range workflows {
result[i] = WorkflowWithSteps{WorkflowDefinition: wf, Steps: []model.WorkflowStep{}}
model.DB.Where("workflow_id = ?", wf.ID).Order("step_number").Find(&result[i].Steps)
}
c.JSON(http.StatusOK, result)
}
// GET /api/v1/instance-groups
func ListInstanceGroups(c *gin.Context) {
var groups []model.InstanceGroup
model.DB.Order("id").Find(&groups)
result := make([]gin.H, len(groups))
for i, g := range groups {
result[i] = enrichGroup(g)
}
c.JSON(http.StatusOK, result)
}
// POST /api/v1/notifications/test
func TestNotification(c *gin.Context) {
_ = time.Now()
c.JSON(http.StatusOK, gin.H{"message": "Notification sent (placeholder)"})
}
// ==================== Audit Rules ====================
// defaultAuditRules defines seed rules per db_type
var defaultAuditRules = map[string][]struct {
Name string
Desc string
RuleType string
Severity string
Pattern string
Message string
Value string
}{
"mysql": {
{"禁止 DROP TABLE", "生产环境不允许删除表", "DDL", "error", `(?i)^\s*DROP\s+TABLE`, "禁止执行 DROP TABLE 操作,可能导致数据永久丢失", "reject"},
{"禁止 DROP DATABASE", "不允许删除数据库", "DDL", "error", `(?i)^\s*DROP\s+DATABASE`, "禁止执行 DROP DATABASE 操作", "reject"},
{"禁止 TRUNCATE TABLE", "清空表数据需走审批", "DDL", "error", `(?i)^\s*TRUNCATE\s+TABLE?`, "禁止执行 TRUNCATE 操作,请使用 DELETE 逐批删除", "reject"},
{"ALTER TABLE 需审核", "表结构变更需审批", "DDL", "warning", `(?i)^\s*ALTER\s+TABLE`, "ALTER TABLE 操作需要审批后执行", "review"},
{"禁止 DELETE 无 WHERE", "删除操作必须带条件", "DML", "error", `(?i)^\s*DELETE\s+FROM?\s+\S+\s*$`, "DELETE 语句必须包含 WHERE 条件", "reject"},
{"禁止 UPDATE 无 WHERE", "更新操作必须带条件", "DML", "error", `(?i)^\s*UPDATE\s+\S+\s+SET\s+.*\s*$`, "UPDATE 语句必须包含 WHERE 条件", "reject"},
{"SELECT * 建议", "避免使用 SELECT *", "DQL", "info", `(?i)^\s*SELECT\s+\*\s+FROM`, "建议明确指定查询字段,避免使用 SELECT *", "warn"},
{"INSERT 需指定列", "INSERT 需指定列名", "DML", "warning", `(?i)^\s*INSERT\s+INTO\s+\S+\s+VALUES`, "INSERT 语句应明确指定列名", "review"},
{"禁止 GRANT ALL", "不允许授予全部权限", "DCL", "error", `(?i)^\s*GRANT\s+ALL`, "禁止使用 GRANT ALL PRIVILEGES", "reject"},
{"大表变更警告", "影响行数超过阈值需注意", "performance", "warning", "", "操作影响行数可能较大,建议在低峰期执行", "warn"},
{"禁止不带 LIMIT 的 DELETE", "批量删除建议带 LIMIT", "DML", "warning", `(?i)^\s*DELETE\s+FROM\s+\S+\s+WHERE`, "建议 DELETE 语句添加 LIMIT 子句分批删除", "warn"},
},
"postgresql": {
{"禁止 DROP TABLE", "生产环境不允许删除表", "DDL", "error", `(?i)^\s*DROP\s+TABLE`, "禁止执行 DROP TABLE 操作", "reject"},
{"禁止 DROP DATABASE", "不允许删除数据库", "DDL", "error", `(?i)^\s*DROP\s+DATABASE`, "禁止执行 DROP DATABASE 操作", "reject"},
{"禁止 DELETE 无 WHERE", "删除操作必须带条件", "DML", "error", `(?i)^\s*DELETE\s+FROM?\s+\S+\s*$`, "DELETE 语句必须包含 WHERE 条件", "reject"},
{"禁止 UPDATE 无 WHERE", "更新操作必须带条件", "DML", "error", `(?i)^\s*UPDATE\s+\S+\s+SET\s+.*\s*$`, "UPDATE 语句必须包含 WHERE 条件", "reject"},
{"ALTER TABLE 需审核", "表结构变更需审批", "DDL", "warning", `(?i)^\s*ALTER\s+TABLE`, "ALTER TABLE 操作需要审批后执行", "review"},
{"SELECT * 建议", "避免使用 SELECT *", "DQL", "info", `(?i)^\s*SELECT\s+\*\s+FROM`, "建议明确指定查询字段", "warn"},
{"禁止 GRANT ALL", "不允许授予全部权限", "DCL", "error", `(?i)^\s*GRANT\s+ALL`, "禁止使用 GRANT ALL", "reject"},
},
"redis": {
{"禁止 FLUSHALL", "不允许清空所有数据", "DDL", "error", `(?i)^FLUSHALL`, "禁止执行 FLUSHALL 操作", "reject"},
{"禁止 FLUSHDB", "清空数据库需审批", "DDL", "error", `(?i)^FLUSHDB`, "禁止执行 FLUSHDB 操作", "reject"},
{"禁止 CONFIG", "配置修改需走审批", "security", "error", `(?i)^CONFIG\s`, "CONFIG 命令需要审批后执行", "reject"},
{"KEYS 命令警告", "生产环境禁止 KEYS", "performance", "error", `(?i)^KEYS\s`, "生产环境禁止使用 KEYS 命令,请使用 SCAN", "reject"},
},
"mongodb": {
{"禁止 dropDatabase", "不允许删除数据库", "DDL", "error", `dropDatabase`, "禁止执行 dropDatabase 操作", "reject"},
{"禁止 drop 集合", "不允许删除集合", "DDL", "error", `(?i)\.drop\(\)`, "禁止执行集合 drop 操作", "reject"},
{"remove 需带条件", "删除操作必须带条件", "DML", "error", `(?i)remove\(\s*\)`, "remove 操作必须带查询条件", "reject"},
{"updateMany 需审核", "批量更新需审批", "DML", "warning", `(?i)updateMany`, "updateMany 操作需要审批后执行", "review"},
},
}
// POST /api/v1/audit-rules/seed/:dbType
func SeedAuditRules(c *gin.Context) {
dbType := c.Param("dbType")
rules, ok := defaultAuditRules[dbType]
if !ok {
resp.BadRequest(c, fmt.Sprintf("不支持的数据库类型: %s", dbType))
return
}
enabled := true
count := 0
for _, r := range rules {
// Idempotent: skip if rule with same name+db_type exists
var existing int64
model.DB.Model(&model.AuditRule{}).Where("name = ? AND db_type = ?", r.Name, dbType).Count(&existing)
if existing > 0 {
continue
}
desc := r.Desc
pattern := r.Pattern
severity := r.Severity
value := r.Value
rule := model.AuditRule{
Name: r.Name,
Desc: &desc,
RuleType: r.RuleType,
Severity: &severity,
DBType: &dbType,
Pattern: &pattern,
IsEnabled: &enabled,
Message: r.Message,
Value: &value,
}
if err := model.DB.Create(&rule).Error; err != nil {
resp.InternalErr(c, err)
return
}
count++
}
c.JSON(http.StatusOK, gin.H{"message": fmt.Sprintf("成功初始化 %d 条规则", count), "count": count})
}
// POST /api/v1/audit-rules/:id/toggle
func ToggleAuditRule(c *gin.Context) {
id, err := strconv.Atoi(c.Param("id"))
if err != nil {
resp.BadRequest(c, "无效的规则 ID")
return
}
var rule model.AuditRule
if err := model.DB.First(&rule, id).Error; err != nil {
resp.NotFound(c, "规则不存在")
return
}
newEnabled := !*rule.IsEnabled
model.DB.Model(&rule).Update("is_enabled", newEnabled)
c.JSON(http.StatusOK, gin.H{"id": id, "is_enabled": newEnabled})
}
// PUT /api/v1/audit-rules/:id
func UpdateAuditRule(c *gin.Context) {
id, err := strconv.Atoi(c.Param("id"))
if err != nil {
resp.BadRequest(c, "无效的规则 ID")
return
}
var rule model.AuditRule
if err := model.DB.First(&rule, id).Error; err != nil {
resp.NotFound(c, "规则不存在")
return
}
var body struct {
Value *string `json:"value"`
Pattern *string `json:"pattern"`
Message *string `json:"message"`
Severity *string `json:"severity"`
}
if err := c.ShouldBindJSON(&body); err != nil {
resp.BadRequest(c, err.Error())
return
}
updates := map[string]interface{}{}
if body.Value != nil {
updates["value"] = *body.Value
}
if body.Pattern != nil {
updates["pattern"] = *body.Pattern
}
if body.Message != nil {
updates["message"] = *body.Message
}
if body.Severity != nil {
updates["severity"] = *body.Severity
}
if len(updates) > 0 {
model.DB.Model(&rule).Updates(updates)
}
c.JSON(http.StatusOK, rule)
}