opensource/SamWaf
3
0
Fork 0
mirror of https://gitee.com/samwaf/SamWaf.git synced 2026-10-31 02:01:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: ip tag db chose

Browse Source 
#505
This commit is contained in:
samwaf
2025-10-30 16:56:24 +08:00
parent 7d24a916ae
commit 458b810dd1
9 changed files with 451 additions and 399 deletions
Download Patch File Download Diff File Expand all files Collapse all files
global/global.go
+15 -5
View File
@@ -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 // 默认使用主数据库
}
service/waf_service/waf_ip_tag_log.go
+158
View File
@@ -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
}
service/waf_service/waf_log.go
-149
View File
@@ -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
}
/*
*
判断是否合法
service/waf_service/waf_stat.go
+3 -2
View File
@@ -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{
wafdb/db_monitor.go
+251
View File
@@ -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)
}
wafdb/localdb.go
+2 -240
View File
@@ -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)
waftask/stat_collector.go
+4 -3
View File
@@ -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,
waftask/task_config.go
+4
View File
@@ -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)
}
waftask/task_createindex.go
+14
View File
@@ -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())