From 458b810dd19ae2f3b7ac8d122e6162602b7609e7 Mon Sep 17 00:00:00 2001 From: samwaf Date: Thu, 30 Oct 2025 16:56:24 +0800 Subject: [PATCH] feat: ip tag db chose #505 --- global/global.go | 20 +- service/waf_service/waf_ip_tag_log.go | 158 ++++++++++++++++ service/waf_service/waf_log.go | 149 --------------- service/waf_service/waf_stat.go | 5 +- wafdb/db_monitor.go | 251 ++++++++++++++++++++++++++ wafdb/localdb.go | 242 +------------------------ waftask/stat_collector.go | 7 +- waftask/task_config.go | 4 + waftask/task_createindex.go | 14 ++ 9 files changed, 451 insertions(+), 399 deletions(-) create mode 100644 service/waf_service/waf_ip_tag_log.go create mode 100644 wafdb/db_monitor.go diff --git a/global/global.go b/global/global.go index 8282b57..51d5dde 100644 --- a/global/global.go +++ b/global/global.go @@ -142,11 +142,13 @@ var ( GSSL_HTTP_CHANGLE_PATH string = "/.well-known/acme-challenge/" // http01证书验证路径 /******数据库处理参数*****/ - GDATA_BATCH_INSERT int64 = 100 //最大批量插入数量 - GDATA_SHARE_DB_SIZE int64 = 100 * 10000 //100w 进行分库 100*10000 - GDATA_SHARE_DB_FILE_SIZE int64 = 1024 //1024M 进行分库 - GDATA_CURRENT_CHANGE bool = false //当前是否正在切换 - GDATA_CURRENT_LOG_DB_MAP map[string]*gorm.DB //当前自定义的数据库连接 TODO 如果用户打开了多个 会不会影响内存 + GDATA_BATCH_INSERT int64 = 100 //最大批量插入数量 + GDATA_SHARE_DB_SIZE int64 = 100 * 10000 //100w 进行分库 100*10000 + GDATA_SHARE_DB_FILE_SIZE int64 = 1024 //1024M 进行分库 + GDATA_CURRENT_CHANGE bool = false //当前是否正在切换 + GDATA_IP_TAG_DB int64 = 0 //IP Tag 存放位置 0 是主库 1是读取 stat库 + + GDATA_CURRENT_LOG_DB_MAP map[string]*gorm.DB //当前自定义的数据库连接 TODO 如果用户打开了多个 会不会影响内存 /******WebSocket*********/ GWebSocket *gwebsocket.WebSocketOnline @@ -286,3 +288,11 @@ func UpdateRealtimeQPS() { atomic.StoreUint64(&GWAF_LAST_LOG_QPS_VALUE, currentLogQPS) atomic.StoreInt64(&GWAF_LAST_QPS_TIME, currentTime) } + +// GetIPTagDB 根据配置获取IP标签数据库连接 +func GetIPTagDB() *gorm.DB { + if GDATA_IP_TAG_DB == 1 { + return GWAF_LOCAL_STATS_DB // 使用统计数据库 + } + return GWAF_LOCAL_DB // 默认使用主数据库 +} diff --git a/service/waf_service/waf_ip_tag_log.go b/service/waf_service/waf_ip_tag_log.go new file mode 100644 index 0000000..9adbe81 --- /dev/null +++ b/service/waf_service/waf_ip_tag_log.go @@ -0,0 +1,158 @@ +package waf_service + +import ( + "SamWaf/global" + "SamWaf/model" + "SamWaf/model/request" + "fmt" + "time" +) + +// GetAttackIpListApi 访问IP列表 +func (receiver *WafLogService) GetAttackIpListApi(req request.WafAttackIpTagSearch) ([]model.AttackIPTag, int64, error) { + var results []model.AttackIPTag + var total int64 + + // 获取本地时区偏移量（秒） + _, offset := time.Now().Zone() + offsetMinutes := offset / 60 + + // 构建时区偏移修饰符 + var offsetModifier string + if offsetMinutes >= 0 { + offsetModifier = fmt.Sprintf("'+%d minutes'", offsetMinutes) + } else { + offsetModifier = fmt.Sprintf("'%d minutes'", offsetMinutes) // 负数自带负号 + } + + // 基础查询部分 + query := ` + SELECT + tenant_id, + user_code, + ip, + SUM(CASE WHEN ip_tag = '正常' THEN cnt ELSE 0 END) AS pass_num, + SUM(CASE WHEN ip_tag <> '正常' THEN cnt ELSE 0 END) AS deny_num, + strftime('%Y-%m-%d %H:%M:%S', MIN(update_time), ` + offsetModifier + `) AS first_time, + strftime('%Y-%m-%d %H:%M:%S', MAX(update_time), ` + offsetModifier + `) AS latest_time, + GROUP_CONCAT(DISTINCT CASE WHEN ip_tag <> '正常' THEN ip_tag END) AS ip_total_tag + FROM + ip_tags + WHERE tenant_id=? and user_code=?` + + // 动态添加过滤条件 + if req.Rule != "" { + query += " AND ip_tag = ?" + } + if req.SrcIp != "" { + query += " AND ip = ?" + } + + // 完成查询的其他部分 + query += ` + GROUP BY + tenant_id, + user_code, + ip + HAVING + SUM(CASE WHEN ip_tag <> '正常' THEN cnt ELSE 0 END) > 0 + ORDER BY + MAX(update_time) DESC + LIMIT ? OFFSET ?` + + // 构建查询参数 + params := []interface{}{global.GWAF_TENANT_ID, global.GWAF_USER_CODE} + + // 添加 Rule 和 SrcIp 作为参数（如果提供了） + if req.Rule != "" { + params = append(params, req.Rule) + } + if req.SrcIp != "" { + params = append(params, req.SrcIp) + } + + // 分页参数 + params = append(params, req.PageSize, req.PageSize*(req.PageIndex-1)) + + // 执行查询 + ipTagDB := global.GetIPTagDB() // 使用封装方法获取数据库连接 + if err := ipTagDB.Raw(query, params...).Scan(&results).Error; err != nil { + return nil, 0, err + } + + // 获取总记录数 + countQuery := ` + SELECT + COUNT(*) AS total + FROM ( + SELECT + tenant_id, + user_code, + ip + FROM + ip_tags + WHERE tenant_id=? and user_code=?` + + // 动态添加过滤条件 + if req.Rule != "" { + countQuery += " AND ip_tag = ?" + } + if req.SrcIp != "" { + countQuery += " AND ip = ?" + } + + countQuery += ` + GROUP BY + tenant_id, + user_code, + ip + HAVING + SUM(CASE WHEN ip_tag <> '正常' THEN cnt ELSE 0 END) > 0 + ) AS subquery` + + // 获取总记录数参数 + countParams := []interface{}{global.GWAF_TENANT_ID, global.GWAF_USER_CODE} + if req.Rule != "" { + countParams = append(countParams, req.Rule) + } + if req.SrcIp != "" { + countParams = append(countParams, req.SrcIp) + } + + // 执行记录数查询 + if err := ipTagDB.Raw(countQuery, countParams...).Scan(&total).Error; err != nil { + return nil, 0, err + } + + return results, total, nil +} + +// GetAllAttackIPTagListApi 获取所有攻击Tag +func (receiver *WafLogService) GetAllAttackIPTagListApi() ([]model.AllIPTag, error) { + var results []model.AllIPTag + + // 基础查询部分 + query := ` +SELECT + ip_tag as value, + ip_tag || ' (' || sum(cnt) || ')' as label + FROM + "ip_tags" +WHERE ip_tag<>'正常' and tenant_id=? and user_code=? + GROUP BY + tenant_id, + ip_tag +order by sum(cnt) desc +` + + // 构建查询参数 + params := []interface{}{global.GWAF_TENANT_ID, global.GWAF_USER_CODE} + + // 执行查询 + ipTagDB := global.GetIPTagDB() // 使用封装方法获取数据库连接 + if err := ipTagDB.Raw(query, params...).Scan(&results).Error; err != nil { + return nil, err + } + + return results, nil +} diff --git a/service/waf_service/waf_log.go b/service/waf_service/waf_log.go index 32e38cb..012dd31 100644 --- a/service/waf_service/waf_log.go +++ b/service/waf_service/waf_log.go @@ -4,14 +4,12 @@ import ( "SamWaf/common/validfield" "SamWaf/global" "SamWaf/innerbean" - "SamWaf/model" "SamWaf/model/request" "SamWaf/wafdb" "errors" "fmt" "strconv" "strings" - "time" ) type WafLogService struct{} @@ -217,153 +215,6 @@ func (receiver *WafLogService) GetUnixTimeByCounter(lastStartCreateUnix int64, l return weblog } -// GetAttackIpListApi 访问IP列表 -func (receiver *WafLogService) GetAttackIpListApi(req request.WafAttackIpTagSearch) ([]model.AttackIPTag, int64, error) { - var results []model.AttackIPTag - var total int64 - - // 获取本地时区偏移量（秒） - _, offset := time.Now().Zone() - offsetMinutes := offset / 60 - - // 构建时区偏移修饰符 - var offsetModifier string - if offsetMinutes >= 0 { - offsetModifier = fmt.Sprintf("'+%d minutes'", offsetMinutes) - } else { - offsetModifier = fmt.Sprintf("'%d minutes'", offsetMinutes) // 负数自带负号 - } - - // 基础查询部分 - query := ` - SELECT - tenant_id, - user_code, - ip, - SUM(CASE WHEN ip_tag = '正常' THEN cnt ELSE 0 END) AS pass_num, - SUM(CASE WHEN ip_tag <> '正常' THEN cnt ELSE 0 END) AS deny_num, - strftime('%Y-%m-%d %H:%M:%S', MIN(update_time), ` + offsetModifier + `) AS first_time, - strftime('%Y-%m-%d %H:%M:%S', MAX(update_time), ` + offsetModifier + `) AS latest_time, - GROUP_CONCAT(DISTINCT CASE WHEN ip_tag <> '正常' THEN ip_tag END) AS ip_total_tag - FROM - ip_tags - WHERE tenant_id=? and user_code=?` - - // 动态添加过滤条件 - if req.Rule != "" { - query += " AND ip_tag = ?" - } - if req.SrcIp != "" { - query += " AND ip = ?" - } - - // 完成查询的其他部分 - query += ` - GROUP BY - tenant_id, - user_code, - ip - HAVING - SUM(CASE WHEN ip_tag <> '正常' THEN cnt ELSE 0 END) > 0 - ORDER BY - MAX(update_time) DESC - LIMIT ? OFFSET ?` - - // 构建查询参数 - params := []interface{}{global.GWAF_TENANT_ID, global.GWAF_USER_CODE} - - // 添加 Rule 和 SrcIp 作为参数（如果提供了） - if req.Rule != "" { - params = append(params, req.Rule) - } - if req.SrcIp != "" { - params = append(params, req.SrcIp) - } - - // 分页参数 - params = append(params, req.PageSize, req.PageSize*(req.PageIndex-1)) - - // 执行查询 - if err := global.GWAF_LOCAL_DB.Raw(query, params...).Scan(&results).Error; err != nil { - return nil, 0, err - } - - // 获取总记录数 - countQuery := ` - SELECT - COUNT(*) AS total - FROM ( - SELECT - tenant_id, - user_code, - ip - FROM - ip_tags - WHERE tenant_id=? and user_code=?` - - // 动态添加过滤条件 - if req.Rule != "" { - countQuery += " AND ip_tag = ?" - } - if req.SrcIp != "" { - countQuery += " AND ip = ?" - } - - countQuery += ` - GROUP BY - tenant_id, - user_code, - ip - HAVING - SUM(CASE WHEN ip_tag <> '正常' THEN cnt ELSE 0 END) > 0 - ) AS subquery` - - // 获取总记录数参数 - countParams := []interface{}{global.GWAF_TENANT_ID, global.GWAF_USER_CODE} - if req.Rule != "" { - countParams = append(countParams, req.Rule) - } - if req.SrcIp != "" { - countParams = append(countParams, req.SrcIp) - } - - // 执行记录数查询 - if err := global.GWAF_LOCAL_DB.Raw(countQuery, countParams...).Scan(&total).Error; err != nil { - return nil, 0, err - } - - return results, total, nil -} - -// GetAllAttackIPTagListApi 获取所有攻击Tag -func (receiver *WafLogService) GetAllAttackIPTagListApi() ([]model.AllIPTag, error) { - var results []model.AllIPTag - - // 基础查询部分 - query := ` -SELECT - ip_tag as value, - ip_tag || ' (' || sum(cnt) || ')' as label - FROM - "ip_tags" -WHERE ip_tag<>'正常' and tenant_id=? and user_code=? - GROUP BY - tenant_id, - ip_tag -order by sum(cnt) desc -` - - // 构建查询参数 - params := []interface{}{global.GWAF_TENANT_ID, global.GWAF_USER_CODE} - - // 执行查询 - if err := global.GWAF_LOCAL_DB.Raw(query, params...).Scan(&results).Error; err != nil { - return nil, err - } - - return results, nil -} - /* * 判断是否合法 diff --git a/service/waf_service/waf_stat.go b/service/waf_service/waf_stat.go index 7f1bcdc..c67395c 100644 --- a/service/waf_service/waf_stat.go +++ b/service/waf_service/waf_stat.go @@ -112,11 +112,12 @@ func (receiver *WafStatService) StatHomeSumDayTopIPRangeApi(req request.WafStats Scan(&AttackCountOfRange) var AttackCountOfRangeMore []model.StatsIPCountMore + ipTagDB := global.GetIPTagDB() // 使用封装方法获取数据库连接 for i := range AttackCountOfRange { region := utils.GetCountry(AttackCountOfRange[i].IP) //查询IP标签 var ipTags []model.IPTag - global.GWAF_LOCAL_DB.Where("tenant_id = ? and user_code = ? and ip=?", + ipTagDB.Where("tenant_id = ? and user_code = ? and ip=?", global.GWAF_TENANT_ID, global.GWAF_USER_CODE, AttackCountOfRange[i].IP).Find(&ipTags) statMore := model.StatsIPCountMore{ @@ -143,7 +144,7 @@ func (receiver *WafStatService) StatHomeSumDayTopIPRangeApi(req request.WafStats //查询IP标签 var ipTags []model.IPTag - global.GWAF_LOCAL_DB.Where("tenant_id = ? and user_code = ? and ip=?", + ipTagDB.Where("tenant_id = ? and user_code = ? and ip=?", global.GWAF_TENANT_ID, global.GWAF_USER_CODE, NormalCountOfRange[i].IP).Find(&ipTags) statMore := model.StatsIPCountMore{ diff --git a/wafdb/db_monitor.go b/wafdb/db_monitor.go new file mode 100644 index 0000000..0e23d0d --- /dev/null +++ b/wafdb/db_monitor.go @@ -0,0 +1,251 @@ +package wafdb + +import ( + "SamWaf/common/zlog" + "SamWaf/global" + "SamWaf/utils" + "encoding/json" + "fmt" + "gorm.io/gorm" + "os" + "time" +) + +// DatabaseMetrics 数据库性能指标结构体 +type DatabaseMetrics struct { + DatabaseName string `json:"database_name"` // 数据库名称 + DatabasePath string `json:"database_path"` // 数据库文件路径 + FileSize int64 `json:"file_size"` // 文件大小(字节) + FileSizeMB float64 `json:"file_size_mb"` // 文件大小(MB) + ConnectionCount int `json:"connection_count"` // 连接数 + MaxConnections int `json:"max_connections"` // 最大连接数 + IdleConnections int `json:"idle_connections"` // 空闲连接数 + InUseConnections int `json:"in_use_connections"` // 使用中连接数 + + // SQLite特有的PRAGMA信息 + PageCount int64 `json:"page_count"` // 页面总数 + PageSize int64 `json:"page_size"` // 页面大小 + FreelistCount int64 `json:"freelist_count"` // 空闲页面数 + CacheSize int64 `json:"cache_size"` // 缓存大小 + JournalMode string `json:"journal_mode"` // 日志模式 + SynchronousMode string `json:"synchronous_mode"` // 同步模式 + TempStore string `json:"temp_store"` // 临时存储模式 + + // 性能统计 + QueryCount int64 `json:"query_count"` // 查询次数(估算) + WriteCount int64 `json:"write_count"` // 写入次数(估算) + CacheHitRatio float64 `json:"cache_hit_ratio"` // 缓存命中率 + + // 时间戳 + Timestamp time.Time `json:"timestamp"` // 采集时间 +} + +// GetDatabaseMetrics 获取指定数据库的性能指标 +func GetDatabaseMetrics(db *gorm.DB, dbName string, dbPath string) (*DatabaseMetrics, error) { + if db == nil { + return nil, fmt.Errorf("数据库连接为空: %s", dbName) + } + + metrics := &DatabaseMetrics{ + DatabaseName: dbName, + DatabasePath: dbPath, + Timestamp: time.Now(), + } + + // 获取底层sql.DB + sqlDB, err := db.DB() + if err != nil { + return nil, fmt.Errorf("获取sql.DB失败: %v", err) + } + + // 获取连接池统计信息 + stats := sqlDB.Stats() + metrics.ConnectionCount = stats.OpenConnections + metrics.MaxConnections = stats.MaxOpenConnections + metrics.IdleConnections = stats.Idle + metrics.InUseConnections = stats.InUse + + // 获取文件大小 + if fileInfo, err := os.Stat(dbPath); err == nil { + metrics.FileSize = fileInfo.Size() + metrics.FileSizeMB = float64(fileInfo.Size()) / (1024 * 1024) + } + + // 获取SQLite PRAGMA信息 + var pageCount, pageSize, freelistCount, cacheSize int64 + var journalMode, synchronousMode, tempStore string + + // 页面信息 + db.Raw("PRAGMA page_count").Scan(&pageCount) + db.Raw("PRAGMA page_size").Scan(&pageSize) + db.Raw("PRAGMA freelist_count").Scan(&freelistCount) + db.Raw("PRAGMA cache_size").Scan(&cacheSize) + + // 模式信息 + db.Raw("PRAGMA journal_mode").Scan(&journalMode) + db.Raw("PRAGMA synchronous").Scan(&synchronousMode) + db.Raw("PRAGMA temp_store").Scan(&tempStore) + + metrics.PageCount = pageCount + metrics.PageSize = pageSize + metrics.FreelistCount = freelistCount + metrics.CacheSize = cacheSize + metrics.JournalMode = journalMode + metrics.SynchronousMode = synchronousMode + metrics.TempStore = tempStore + + // 获取缓存命中率等统计信息 + var cacheHit, cacheMiss int64 + db.Raw("PRAGMA cache_hit").Scan(&cacheHit) + db.Raw("PRAGMA cache_miss").Scan(&cacheMiss) + + if cacheHit+cacheMiss > 0 { + metrics.CacheHitRatio = float64(cacheHit) / float64(cacheHit+cacheMiss) * 100 + } + + return metrics, nil +} + +// MonitorAllDatabases 监控所有数据库的性能指标 +func MonitorAllDatabases() ([]*DatabaseMetrics, error) { + var allMetrics []*DatabaseMetrics + currentDir := utils.GetCurrentDir() + + // 监控主数据库 + if global.GWAF_LOCAL_DB != nil { + mainDbPath := currentDir + "/data/local.db" + if metrics, err := GetDatabaseMetrics(global.GWAF_LOCAL_DB, "主数据库(Core)", mainDbPath); err == nil { + allMetrics = append(allMetrics, metrics) + } else { + zlog.Error("获取主数据库指标失败:", err) + } + } + + // 监控日志数据库 + if global.GWAF_LOCAL_LOG_DB != nil { + logDbPath := currentDir + "/data/local_log.db" + if metrics, err := GetDatabaseMetrics(global.GWAF_LOCAL_LOG_DB, "日志数据库(Log)", logDbPath); err == nil { + allMetrics = append(allMetrics, metrics) + } else { + zlog.Error("获取日志数据库指标失败:", err) + } + } + + // 监控统计数据库 + if global.GWAF_LOCAL_STATS_DB != nil { + statsDbPath := currentDir + "/data/local_stats.db" + if metrics, err := GetDatabaseMetrics(global.GWAF_LOCAL_STATS_DB, "统计数据库(Stats)", statsDbPath); err == nil { + allMetrics = append(allMetrics, metrics) + } else { + zlog.Error("获取统计数据库指标失败:", err) + } + } + + // 监控自定义日志数据库 + if global.GDATA_CURRENT_LOG_DB_MAP != nil { + for fileName, db := range global.GDATA_CURRENT_LOG_DB_MAP { + if db != nil { + customDbPath := currentDir + "/data/" + fileName + dbName := fmt.Sprintf("自定义日志数据库(%s)", fileName) + if metrics, err := GetDatabaseMetrics(db, dbName, customDbPath); err == nil { + allMetrics = append(allMetrics, metrics) + } else { + zlog.Error("获取自定义数据库指标失败:", fileName, err) + } + } + } + } + + return allMetrics, nil +} + +// PrintDatabaseMetrics 打印数据库性能指标 +func PrintDatabaseMetrics() { + metrics, err := MonitorAllDatabases() + if err != nil { + zlog.Error("监控数据库失败:", err) + return + } + + zlog.Info("==================== 数据库性能监控报告 ====================") + zlog.Info("采集时间:", time.Now().Format("2006-01-02 15:04:05")) + zlog.Info("") + + for _, m := range metrics { + // 构建缓存命中率信息 + cacheHitInfo := "" + if m.CacheHitRatio > 0 { + cacheHitInfo = fmt.Sprintf("\n - 缓存命中率: %.2f%%", m.CacheHitRatio) + } + + dbInfo := fmt.Sprintf(`【%s】 + 数据库路径: %s + 文件大小: %.2f MB (%d 字节) + 连接池信息: + - 当前连接数: %d + - 最大连接数: %d + - 空闲连接数: %d + - 使用中连接数: %d + SQLite配置: + - 页面总数: %d + - 页面大小: %d 字节 + - 空闲页面数: %d + - 缓存大小: %d + - 日志模式: %s + - 同步模式: %s + - 临时存储: %s%s`, + m.DatabaseName, + m.DatabasePath, + m.FileSizeMB, m.FileSize, + m.ConnectionCount, + m.MaxConnections, + m.IdleConnections, + m.InUseConnections, + m.PageCount, + m.PageSize, + m.FreelistCount, + m.CacheSize, + m.JournalMode, + m.SynchronousMode, + m.TempStore, + cacheHitInfo) + + zlog.Info(dbInfo) + zlog.Info("") + } + zlog.Info("========================================================") +} + +// GetDatabaseMetricsJSON 获取数据库指标的JSON格式 +func GetDatabaseMetricsJSON() (string, error) { + metrics, err := MonitorAllDatabases() + if err != nil { + return "", err + } + + jsonData, err := json.Marshal(metrics) + if err != nil { + return "", fmt.Errorf("序列化JSON失败: %v", err) + } + + return string(jsonData), nil +} + +// StartDatabaseMonitoring 启动定时数据库监控 +func StartDatabaseMonitoring(intervalMinutes int) { + if intervalMinutes <= 0 { + intervalMinutes = 5 // 默认5分钟 + } + + ticker := time.NewTicker(time.Duration(intervalMinutes) * time.Minute) + go func() { + for { + select { + case <-ticker.C: + PrintDatabaseMetrics() + } + } + }() + + zlog.Info("数据库性能监控已启动，监控间隔: %d 分钟", intervalMinutes) +} diff --git a/wafdb/localdb.go b/wafdb/localdb.go index 6b46be4..6f43cb8 100644 --- a/wafdb/localdb.go +++ b/wafdb/localdb.go @@ -11,7 +11,6 @@ import ( "SamWaf/utils" "context" "database/sql" - "encoding/json" "fmt" "gorm.io/gorm/logger" "net/url" @@ -175,245 +174,6 @@ func InitCoreDb(currentDir string) (bool, error) { } } -// DatabaseMetrics 数据库性能指标结构体 -type DatabaseMetrics struct { - DatabaseName string `json:"database_name"` // 数据库名称 - DatabasePath string `json:"database_path"` // 数据库文件路径 - FileSize int64 `json:"file_size"` // 文件大小(字节) - FileSizeMB float64 `json:"file_size_mb"` // 文件大小(MB) - ConnectionCount int `json:"connection_count"` // 连接数 - MaxConnections int `json:"max_connections"` // 最大连接数 - IdleConnections int `json:"idle_connections"` // 空闲连接数 - InUseConnections int `json:"in_use_connections"` // 使用中连接数 - - // SQLite特有的PRAGMA信息 - PageCount int64 `json:"page_count"` // 页面总数 - PageSize int64 `json:"page_size"` // 页面大小 - FreelistCount int64 `json:"freelist_count"` // 空闲页面数 - CacheSize int64 `json:"cache_size"` // 缓存大小 - JournalMode string `json:"journal_mode"` // 日志模式 - SynchronousMode string `json:"synchronous_mode"` // 同步模式 - TempStore string `json:"temp_store"` // 临时存储模式 - - // 性能统计 - QueryCount int64 `json:"query_count"` // 查询次数(估算) - WriteCount int64 `json:"write_count"` // 写入次数(估算) - CacheHitRatio float64 `json:"cache_hit_ratio"` // 缓存命中率 - - // 时间戳 - Timestamp time.Time `json:"timestamp"` // 采集时间 -} - -// GetDatabaseMetrics 获取指定数据库的性能指标 -func GetDatabaseMetrics(db *gorm.DB, dbName string, dbPath string) (*DatabaseMetrics, error) { - if db == nil { - return nil, fmt.Errorf("数据库连接为空: %s", dbName) - } - - metrics := &DatabaseMetrics{ - DatabaseName: dbName, - DatabasePath: dbPath, - Timestamp: time.Now(), - } - - // 获取底层sql.DB - sqlDB, err := db.DB() - if err != nil { - return nil, fmt.Errorf("获取sql.DB失败: %v", err) - } - - // 获取连接池统计信息 - stats := sqlDB.Stats() - metrics.ConnectionCount = stats.OpenConnections - metrics.MaxConnections = stats.MaxOpenConnections - metrics.IdleConnections = stats.Idle - metrics.InUseConnections = stats.InUse - - // 获取文件大小 - if fileInfo, err := os.Stat(dbPath); err == nil { - metrics.FileSize = fileInfo.Size() - metrics.FileSizeMB = float64(fileInfo.Size()) / (1024 * 1024) - } - - // 获取SQLite PRAGMA信息 - var pageCount, pageSize, freelistCount, cacheSize int64 - var journalMode, synchronousMode, tempStore string - - // 页面信息 - db.Raw("PRAGMA page_count").Scan(&pageCount) - db.Raw("PRAGMA page_size").Scan(&pageSize) - db.Raw("PRAGMA freelist_count").Scan(&freelistCount) - db.Raw("PRAGMA cache_size").Scan(&cacheSize) - - // 模式信息 - db.Raw("PRAGMA journal_mode").Scan(&journalMode) - db.Raw("PRAGMA synchronous").Scan(&synchronousMode) - db.Raw("PRAGMA temp_store").Scan(&tempStore) - - metrics.PageCount = pageCount - metrics.PageSize = pageSize - metrics.FreelistCount = freelistCount - metrics.CacheSize = cacheSize - metrics.JournalMode = journalMode - metrics.SynchronousMode = synchronousMode - metrics.TempStore = tempStore - - // 获取缓存命中率等统计信息 - var cacheHit, cacheMiss int64 - db.Raw("PRAGMA cache_hit").Scan(&cacheHit) - db.Raw("PRAGMA cache_miss").Scan(&cacheMiss) - - if cacheHit+cacheMiss > 0 { - metrics.CacheHitRatio = float64(cacheHit) / float64(cacheHit+cacheMiss) * 100 - } - - return metrics, nil -} - -// MonitorAllDatabases 监控所有数据库的性能指标 -func MonitorAllDatabases() ([]*DatabaseMetrics, error) { - var allMetrics []*DatabaseMetrics - currentDir := utils.GetCurrentDir() - - // 监控主数据库 - if global.GWAF_LOCAL_DB != nil { - mainDbPath := currentDir + "/data/local.db" - if metrics, err := GetDatabaseMetrics(global.GWAF_LOCAL_DB, "主数据库(Core)", mainDbPath); err == nil { - allMetrics = append(allMetrics, metrics) - } else { - zlog.Error("获取主数据库指标失败:", err) - } - } - - // 监控日志数据库 - if global.GWAF_LOCAL_LOG_DB != nil { - logDbPath := currentDir + "/data/local_log.db" - if metrics, err := GetDatabaseMetrics(global.GWAF_LOCAL_LOG_DB, "日志数据库(Log)", logDbPath); err == nil { - allMetrics = append(allMetrics, metrics) - } else { - zlog.Error("获取日志数据库指标失败:", err) - } - } - - // 监控统计数据库 - if global.GWAF_LOCAL_STATS_DB != nil { - statsDbPath := currentDir + "/data/local_stats.db" - if metrics, err := GetDatabaseMetrics(global.GWAF_LOCAL_STATS_DB, "统计数据库(Stats)", statsDbPath); err == nil { - allMetrics = append(allMetrics, metrics) - } else { - zlog.Error("获取统计数据库指标失败:", err) - } - } - - // 监控自定义日志数据库 - if global.GDATA_CURRENT_LOG_DB_MAP != nil { - for fileName, db := range global.GDATA_CURRENT_LOG_DB_MAP { - if db != nil { - customDbPath := currentDir + "/data/" + fileName - dbName := fmt.Sprintf("自定义日志数据库(%s)", fileName) - if metrics, err := GetDatabaseMetrics(db, dbName, customDbPath); err == nil { - allMetrics = append(allMetrics, metrics) - } else { - zlog.Error("获取自定义数据库指标失败:", fileName, err) - } - } - } - } - - return allMetrics, nil -} - -// PrintDatabaseMetrics 打印数据库性能指标 -func PrintDatabaseMetrics() { - metrics, err := MonitorAllDatabases() - if err != nil { - zlog.Error("监控数据库失败:", err) - return - } - - zlog.Info("==================== 数据库性能监控报告 ====================") - zlog.Info("采集时间:", time.Now().Format("2006-01-02 15:04:05")) - zlog.Info("") - - for _, m := range metrics { - // 构建缓存命中率信息 - cacheHitInfo := "" - if m.CacheHitRatio > 0 { - cacheHitInfo = fmt.Sprintf("\n - 缓存命中率: %.2f%%", m.CacheHitRatio) - } - - dbInfo := fmt.Sprintf(`【%s】 - 数据库路径: %s - 文件大小: %.2f MB (%d 字节) - 连接池信息: - - 当前连接数: %d - - 最大连接数: %d - - 空闲连接数: %d - - 使用中连接数: %d - SQLite配置: - - 页面总数: %d - - 页面大小: %d 字节 - - 空闲页面数: %d - - 缓存大小: %d - - 日志模式: %s - - 同步模式: %s - - 临时存储: %s%s`, - m.DatabaseName, - m.DatabasePath, - m.FileSizeMB, m.FileSize, - m.ConnectionCount, - m.MaxConnections, - m.IdleConnections, - m.InUseConnections, - m.PageCount, - m.PageSize, - m.FreelistCount, - m.CacheSize, - m.JournalMode, - m.SynchronousMode, - m.TempStore, - cacheHitInfo) - - zlog.Info(dbInfo) - zlog.Info("") - } - zlog.Info("========================================================") -} - -// GetDatabaseMetricsJSON 获取数据库指标的JSON格式 -func GetDatabaseMetricsJSON() (string, error) { - metrics, err := MonitorAllDatabases() - if err != nil { - return "", err - } - - jsonData, err := json.Marshal(metrics) - if err != nil { - return "", fmt.Errorf("序列化JSON失败: %v", err) - } - - return string(jsonData), nil -} - -// StartDatabaseMonitoring 启动定时数据库监控 -func StartDatabaseMonitoring(intervalMinutes int) { - if intervalMinutes <= 0 { - intervalMinutes = 5 // 默认5分钟 - } - - ticker := time.NewTicker(time.Duration(intervalMinutes) * time.Minute) - go func() { - for { - select { - case <-ticker.C: - PrintDatabaseMetrics() - } - } - }() - - zlog.Info("数据库性能监控已启动，监控间隔: %d 分钟", intervalMinutes) -} - func InitLogDb(currentDir string) (bool, error) { if currentDir == "" { currentDir = utils.GetCurrentDir() @@ -572,6 +332,8 @@ func InitStatsDb(currentDir string) (bool, error) { db.AutoMigrate(&model.StatsDay{}) db.AutoMigrate(&model.StatsIPDay{}) db.AutoMigrate(&model.StatsIPCityDay{}) + //IPTag + db.AutoMigrate(&model.IPTag{}) global.GWAF_LOCAL_STATS_DB.Callback().Query().Before("gorm:query").Register("tenant_plugin:before_query", before_query) global.GWAF_LOCAL_STATS_DB.Callback().Query().Before("gorm:update").Register("tenant_plugin:before_update", before_update) diff --git a/waftask/stat_collector.go b/waftask/stat_collector.go index c750f77..8705220 100644 --- a/waftask/stat_collector.go +++ b/waftask/stat_collector.go @@ -254,11 +254,12 @@ func CollectStatsFromLogs(logs []*innerbean.WebLog) { } zlog.Debug("城市聚合处理完成", "更新记录数", cityUpdateCount, "插入记录数", cityInsertCount) - // 4) IPTag 增量（主库） + // 4) IPTag 增量（根据配置选择数据库） ipTagUpdateCount := 0 ipTagInsertCount := 0 + ipTagDB := global.GetIPTagDB() // 使用封装方法获取数据库连接 for k, delta := range ipTagAgg { - tx := global.GWAF_LOCAL_DB.Model(&model.IPTag{}). + tx := ipTagDB.Model(&model.IPTag{}). Where("tenant_id = ? and user_code = ? and ip = ? and ip_tag = ?", global.GWAF_TENANT_ID, global.GWAF_USER_CODE, k.IP, k.Rule). Updates(map[string]interface{}{ @@ -270,7 +271,7 @@ func CollectStatsFromLogs(logs []*innerbean.WebLog) { continue } if tx.RowsAffected == 0 { - err := global.GWAF_LOCAL_DB.Create(&model.IPTag{ + err := ipTagDB.Create(&model.IPTag{ BaseOrm: baseorm.BaseOrm{ Id: uuid.GenUUID(), USER_CODE: global.GWAF_USER_CODE, diff --git a/waftask/task_config.go b/waftask/task_config.go index 7124f64..f78def8 100644 --- a/waftask/task_config.go +++ b/waftask/task_config.go @@ -110,6 +110,9 @@ func setConfigIntValue(name string, value int64, change int) { case "enable_system_stats_push": global.GCONFIG_ENABLE_SYSTEM_STATS_PUSH = value break + case "ip_tag_db": + global.GDATA_IP_TAG_DB = value + break default: zlog.Warn("Unknown config item:", name) } @@ -275,4 +278,5 @@ func TaskLoadSetting(initLoad bool) { //数据库相关 updateConfigIntItem(initLoad, "database", "batch_insert", global.GDATA_BATCH_INSERT, "数据库批量插入数量", "int", "", configMap) updateConfigIntItem(initLoad, "database", "log_persist_enable", global.GCONFIG_LOG_PERSIST_ENABLED, "是否开启日志持久化（1开启 0关闭）", "options", "0|关闭,1|开启", configMap) + updateConfigIntItem(initLoad, "database", "ip_tag_db", global.GDATA_IP_TAG_DB, "IP Tag 存放位置 0 是主库 1是读取 stat库", "int", "", configMap) } diff --git a/waftask/task_createindex.go b/waftask/task_createindex.go index dcb865e..1bf6611 100644 --- a/waftask/task_createindex.go +++ b/waftask/task_createindex.go @@ -160,6 +160,20 @@ func createStatDbIndex() { zlog.Info("db", "idx_stats_ip_city_days_lookup created") } + //20251030 创建iptagtag索引 + err = db.Exec("CREATE UNIQUE INDEX IF NOT EXISTS uni_iptags_full ON ip_tags (user_code, tenant_id, ip, ip_tag)").Error + if err != nil { + panic("failed to create index : uni_iptags_full " + err.Error()) + } else { + zlog.Info("db", "uni_iptags_full created") + } + // 创建iptag ip索引 + err = db.Exec("CREATE INDEX IF NOT EXISTS idx_iptag_ip ON ip_tags ( user_code, tenant_id, ip)").Error + if err != nil { + panic("failed to create index: idx_iptag_ip " + err.Error()) + } else { + zlog.Info("db", "idx_iptag_ip created") + } // 记录结束时间并计算耗时 duration := time.Since(startTime) zlog.Info("create stats index completely", "duration", duration.String())