waflog.go 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579
  1. package admin
  2. import (
  3. "context"
  4. "fmt"
  5. "math"
  6. "strings"
  7. "time"
  8. v1 "github.com/go-nunu/nunu-layout-advanced/api/v1"
  9. adminApi "github.com/go-nunu/nunu-layout-advanced/api/v1/admin"
  10. "github.com/go-nunu/nunu-layout-advanced/internal/model"
  11. "github.com/go-nunu/nunu-layout-advanced/internal/repository"
  12. "gorm.io/gorm"
  13. )
  14. type WafLogRepository interface {
  15. GetWafLog(ctx context.Context, id int64) (*model.WafLog, error)
  16. GetWafLogList(ctx context.Context, req adminApi.SearchWafLogParams) (*v1.PaginatedResponse[model.WafLog], error)
  17. AddWafLog(ctx context.Context, log *model.WafLog) error
  18. BatchAddWafLog(ctx context.Context, logs []*model.WafLog) error
  19. ExportWafLog(ctx context.Context, req adminApi.ExportWafLog) ([]model.WafLogWithGatewayIP, error)
  20. ExportWafLogWithPagination(ctx context.Context, req adminApi.ExportWafLog, page, pageSize int) ([]model.WafLogWithGatewayIP, error)
  21. GetWafLogExportCount(ctx context.Context, req adminApi.ExportWafLog) (int, error)
  22. }
  23. func NewWafLogRepository(
  24. repository *repository.Repository,
  25. ) WafLogRepository {
  26. return &wafLogRepository{
  27. Repository: repository,
  28. }
  29. }
  30. type wafLogRepository struct {
  31. *repository.Repository
  32. }
  33. // buildExportQuery 是一个辅助函数,用于构建导出日志的公共查询条件(支持分表)
  34. func (r *wafLogRepository) buildExportQuery(ctx context.Context, req adminApi.ExportWafLog, tableName string) *gorm.DB {
  35. // 使用传入的表名,给表起一个别名,方便子查询中引用
  36. query := r.DBWithName(ctx, "admin").Model(&model.WafLog{}).Table(tableName + " as wl")
  37. if req.RequestIp != "" {
  38. query = query.Where("wl.request_ip = ?", strings.TrimSpace(req.RequestIp))
  39. }
  40. if req.Uid != 0 {
  41. query = query.Where("wl.uid = ?", req.Uid)
  42. }
  43. if req.Api != "" {
  44. query = query.Where("wl.api = ?", strings.TrimSpace(req.Api))
  45. }
  46. if req.Name != "" {
  47. query = query.Where("wl.name = ?", strings.TrimSpace(req.Name))
  48. }
  49. if req.RuleId != 0 {
  50. query = query.Where("wl.rule_id = ?", req.RuleId)
  51. }
  52. if len(req.HostIds) > 0 {
  53. query = query.Where("wl.host_id IN ?", req.HostIds)
  54. }
  55. if req.UserAgent != "" {
  56. query = query.Where("wl.user_agent = ?", strings.TrimSpace(req.UserAgent))
  57. }
  58. if len(req.ApiNames) > 0 {
  59. query = query.Where("wl.api_name IN ?", req.ApiNames)
  60. }
  61. if len(req.ApiTypes) > 0 {
  62. query = query.Where("wl.api_type IN ?", req.ApiTypes)
  63. }
  64. if req.StartTime != "" {
  65. query = query.Where("wl.created_at > ?", strings.TrimSpace(req.StartTime))
  66. }
  67. if req.EndTime != "" {
  68. query = query.Where("wl.created_at < ?", strings.TrimSpace(req.EndTime))
  69. }
  70. return query
  71. }
  72. // getExportTimeRange 根据请求参数解析时间范围,用于确定需要查询的分表
  73. func (r *wafLogRepository) getExportTimeRange(req adminApi.ExportWafLog) (*time.Time, *time.Time) {
  74. var startTime, endTime *time.Time
  75. if req.StartTime != "" {
  76. if t, err := time.Parse("2006-01-02 15:04:05", strings.TrimSpace(req.StartTime)); err == nil {
  77. startTime = &t
  78. } else if t, err := time.Parse("2006-01-02", strings.TrimSpace(req.StartTime)); err == nil {
  79. startTime = &t
  80. }
  81. }
  82. if req.EndTime != "" {
  83. if t, err := time.Parse("2006-01-02 15:04:05", strings.TrimSpace(req.EndTime)); err == nil {
  84. endTime = &t
  85. } else if t, err := time.Parse("2006-01-02", strings.TrimSpace(req.EndTime)); err == nil {
  86. endTime = &t
  87. }
  88. }
  89. return startTime, endTime
  90. }
  91. // buildSubQueryForTable 为指定的表构建子查询
  92. func (r *wafLogRepository) buildSubQueryForTable(ctx context.Context, tableName string) *gorm.DB {
  93. // 需要在所有可能包含"分配网关组"数据的表中查找
  94. // 这里简化处理,先在当前表中查找,如果需要跨表查找可以进一步优化
  95. return r.DBWithName(ctx, "admin").Table(tableName).
  96. Select("extra_data").
  97. Where("api_name = ?", "分配网关组").
  98. Where("host_id = wl.host_id").
  99. Where("uid = wl.uid").
  100. Where("created_at <= wl.created_at").
  101. Order("created_at DESC").
  102. Limit(1)
  103. }
  104. func (r *wafLogRepository) GetWafLog(ctx context.Context, id int64) (*model.WafLog, error) {
  105. var res model.WafLog
  106. // 获取存在的分表
  107. existingTables := r.Manager.GetTableNamesWithExistenceCheck(r.DBWithName(ctx, "admin"), "waf_log", nil, nil)
  108. // 在各个分表中查找
  109. for _, tableName := range existingTables {
  110. err := r.DBWithName(ctx, "admin").Table(tableName).Where("id = ?", id).First(&res).Error
  111. if err == nil {
  112. res.SetTableName(tableName)
  113. return &res, nil
  114. }
  115. }
  116. return nil, fmt.Errorf("未找到ID为 %d 的WAF日志记录", id)
  117. }
  118. func (r *wafLogRepository) GetWafLogList(ctx context.Context, req adminApi.SearchWafLogParams) (*v1.PaginatedResponse[model.WafLog], error) {
  119. //// 解析时间范围(如果有的话)
  120. //var startTime, endTime *time.Time
  121. //// TODO: 这里可以根据req中的时间字段来确定查询范围
  122. //// 暂时查询最近3个月的数据
  123. // 获取需要查询的表
  124. existingTables := r.Manager.GetTableNamesWithExistenceCheck(r.DBWithName(ctx, "admin"), "waf_log", nil, nil)
  125. if len(existingTables) == 0 {
  126. // 没有分表,返回空结果
  127. return &v1.PaginatedResponse[model.WafLog]{
  128. Records: []model.WafLog{},
  129. Page: 1,
  130. PageSize: 10,
  131. Total: 0,
  132. TotalPages: 0,
  133. }, nil
  134. }
  135. if len(existingTables) == 1 {
  136. // 只有一个表,直接查询
  137. return r.queryWafLogFromSingleTable(ctx, req, existingTables[0])
  138. }
  139. // 跨表分页查询
  140. return r.queryWafLogFromMultipleTables(ctx, req, existingTables)
  141. }
  142. // queryWafLogFromSingleTable 单表查询
  143. func (r *wafLogRepository) queryWafLogFromSingleTable(ctx context.Context, req adminApi.SearchWafLogParams, tableName string) (*v1.PaginatedResponse[model.WafLog], error) {
  144. var res []model.WafLog
  145. var total int64
  146. query := r.DBWithName(ctx, "admin").Table(tableName)
  147. query = r.applyWafLogFilters(query, req)
  148. if err := query.Count(&total).Error; err != nil {
  149. return nil, err
  150. }
  151. page := req.Current
  152. pageSize := req.PageSize
  153. if page <= 0 {
  154. page = 1
  155. }
  156. if pageSize <= 0 {
  157. pageSize = 10
  158. } else if pageSize > 100 {
  159. pageSize = 100
  160. }
  161. offset := (page - 1) * pageSize
  162. if req.Column != "" {
  163. query = query.Order(req.Column + " " + req.Order)
  164. }
  165. result := query.Offset(offset).Limit(pageSize).Find(&res)
  166. if result.Error != nil {
  167. return nil, result.Error
  168. }
  169. return &v1.PaginatedResponse[model.WafLog]{
  170. Records: res,
  171. Page: page,
  172. PageSize: pageSize,
  173. Total: total,
  174. TotalPages: int(math.Ceil(float64(total) / float64(pageSize))),
  175. }, nil
  176. }
  177. // queryWafLogFromMultipleTables 多表联合查询
  178. func (r *wafLogRepository) queryWafLogFromMultipleTables(ctx context.Context, req adminApi.SearchWafLogParams, tableNames []string) (*v1.PaginatedResponse[model.WafLog], error) {
  179. var allResults []model.WafLog
  180. var totalCount int64
  181. // 先计算总数
  182. for _, tableName := range tableNames {
  183. var count int64
  184. query := r.DBWithName(ctx, "admin").Table(tableName)
  185. query = r.applyWafLogFilters(query, req)
  186. if err := query.Count(&count).Error; err != nil {
  187. return nil, err
  188. }
  189. totalCount += count
  190. }
  191. page := req.Current
  192. pageSize := req.PageSize
  193. if page <= 0 {
  194. page = 1
  195. }
  196. if pageSize <= 0 {
  197. pageSize = 10
  198. } else if pageSize > 100 {
  199. pageSize = 100
  200. }
  201. // 计算需要跳过的记录数
  202. offset := (page - 1) * pageSize
  203. limit := pageSize
  204. currentOffset := 0
  205. // 逐表查询直到获取足够的记录
  206. for _, tableName := range tableNames {
  207. if limit <= 0 {
  208. break
  209. }
  210. var tableCount int64
  211. countQuery := r.DBWithName(ctx, "admin").Table(tableName)
  212. countQuery = r.applyWafLogFilters(countQuery, req)
  213. if err := countQuery.Count(&tableCount).Error; err != nil {
  214. return nil, err
  215. }
  216. // 如果当前表的记录数不足以满足offset要求,跳过这个表
  217. if currentOffset+int(tableCount) <= offset {
  218. currentOffset += int(tableCount)
  219. continue
  220. }
  221. // 计算在当前表中的offset
  222. tableOffset := offset - currentOffset
  223. if tableOffset < 0 {
  224. tableOffset = 0
  225. }
  226. var tableResults []model.WafLog
  227. query := r.DBWithName(ctx, "admin").Table(tableName)
  228. query = r.applyWafLogFilters(query, req)
  229. if req.Column != "" {
  230. query = query.Order(req.Column + " " + req.Order)
  231. }
  232. err := query.Offset(tableOffset).Limit(limit).Find(&tableResults).Error
  233. if err != nil {
  234. return nil, err
  235. }
  236. // 设置表名
  237. for i := range tableResults {
  238. tableResults[i].SetTableName(tableName)
  239. }
  240. allResults = append(allResults, tableResults...)
  241. limit -= len(tableResults)
  242. currentOffset += int(tableCount)
  243. }
  244. return &v1.PaginatedResponse[model.WafLog]{
  245. Records: allResults,
  246. Page: page,
  247. PageSize: pageSize,
  248. Total: totalCount,
  249. TotalPages: int(math.Ceil(float64(totalCount) / float64(pageSize))),
  250. }, nil
  251. }
  252. func (r *wafLogRepository) AddWafLog(ctx context.Context, log *model.WafLog) error {
  253. // 设置创建时间
  254. if log.CreatedAt.IsZero() {
  255. log.CreatedAt = time.Now()
  256. }
  257. // 获取最优的写入表(自动根据表名获取阈值)
  258. tableName, err := r.Manager.GetOptimalWriteTable(ctx, r.DBWithName(ctx, "admin"), log)
  259. if err != nil {
  260. return fmt.Errorf("获取写入表失败: %v", err)
  261. }
  262. log.SetTableName(tableName)
  263. // 确保表存在
  264. err = r.Manager.EnsureTableExists(ctx, r.DBWithName(ctx, "admin"), tableName, &model.WafLog{})
  265. if err != nil {
  266. return err
  267. }
  268. // 写入数据
  269. return r.DBWithName(ctx, "admin").Table(tableName).Create(log).Error
  270. }
  271. func (r *wafLogRepository) BatchAddWafLog(ctx context.Context, logs []*model.WafLog) error {
  272. if len(logs) == 0 {
  273. return nil
  274. }
  275. // 按表名分组
  276. tableGroups := make(map[string][]*model.WafLog)
  277. for _, log := range logs {
  278. // 设置创建时间
  279. if log.CreatedAt.IsZero() {
  280. log.CreatedAt = time.Now()
  281. }
  282. // 获取最优的写入表(自动根据表名获取阈值)
  283. tableName, err := r.Manager.GetOptimalWriteTable(ctx, r.DBWithName(ctx, "admin"), log)
  284. if err != nil {
  285. return fmt.Errorf("获取写入表失败: %v", err)
  286. }
  287. log.SetTableName(tableName)
  288. // 按表名分组
  289. tableGroups[tableName] = append(tableGroups[tableName], log)
  290. }
  291. // 为每个表批量插入
  292. for tableName, tableLogs := range tableGroups {
  293. // 确保表存在
  294. err := r.Manager.EnsureTableExists(ctx, r.DBWithName(ctx, "admin"), tableName, &model.WafLog{})
  295. if err != nil {
  296. return err
  297. }
  298. // 批量插入
  299. err = r.DBWithName(ctx, "admin").Table(tableName).CreateInBatches(tableLogs, len(tableLogs)).Error
  300. if err != nil {
  301. return err
  302. }
  303. }
  304. return nil
  305. }
  306. func (r *wafLogRepository) ExportWafLog(ctx context.Context, req adminApi.ExportWafLog) ([]model.WafLogWithGatewayIP, error) {
  307. return r.ExportWafLogWithPagination(ctx, req, 0, 0)
  308. }
  309. // ExportWafLogWithPagination 使用子查询获取每条日志在当时时间点的正确网关组IP(支持分表)
  310. func (r *wafLogRepository) ExportWafLogWithPagination(ctx context.Context, req adminApi.ExportWafLog, page, pageSize int) ([]model.WafLogWithGatewayIP, error) {
  311. // 1. 解析时间范围
  312. startTime, endTime := r.getExportTimeRange(req)
  313. // 2. 获取需要查询的分表
  314. existingTables := r.Manager.GetTableNamesWithExistenceCheck(r.DBWithName(ctx, "admin"), "waf_log", startTime, endTime)
  315. if len(existingTables) == 0 {
  316. return []model.WafLogWithGatewayIP{}, nil
  317. }
  318. if len(existingTables) == 1 {
  319. // 单表查询
  320. return r.exportFromSingleTable(ctx, req, existingTables[0], page, pageSize)
  321. }
  322. // 多表查询(不分页,获取所有数据)
  323. if page > 0 && pageSize > 0 {
  324. return r.exportFromMultipleTablesWithPagination(ctx, req, existingTables, page, pageSize)
  325. } else {
  326. return r.exportFromMultipleTables(ctx, req, existingTables)
  327. }
  328. }
  329. // exportFromSingleTable 从单个表导出数据
  330. func (r *wafLogRepository) exportFromSingleTable(ctx context.Context, req adminApi.ExportWafLog, tableName string, page, pageSize int) ([]model.WafLogWithGatewayIP, error) {
  331. var res []model.WafLogWithGatewayIP
  332. // 1. 构建基础查询
  333. query := r.buildExportQuery(ctx, req, tableName)
  334. // 2. 构建子查询
  335. subQuery := r.buildSubQueryForTable(ctx, tableName)
  336. // 3. 添加 Select
  337. query = query.Select("wl.*, (?) as gateway_ip_data", subQuery)
  338. // 4. 添加分页
  339. if page > 0 && pageSize > 0 {
  340. offset := (page - 1) * pageSize
  341. query = query.Offset(offset).Limit(pageSize)
  342. }
  343. // 5. 执行查询
  344. if err := query.Find(&res).Error; err != nil {
  345. if err == gorm.ErrRecordNotFound {
  346. return []model.WafLogWithGatewayIP{}, nil
  347. }
  348. return nil, err
  349. }
  350. return res, nil
  351. }
  352. // exportFromMultipleTables 从多个表导出所有数据(不分页)
  353. func (r *wafLogRepository) exportFromMultipleTables(ctx context.Context, req adminApi.ExportWafLog, tableNames []string) ([]model.WafLogWithGatewayIP, error) {
  354. var allResults []model.WafLogWithGatewayIP
  355. for _, tableName := range tableNames {
  356. tableResults, err := r.exportFromSingleTable(ctx, req, tableName, 0, 0)
  357. if err != nil {
  358. return nil, fmt.Errorf("查询表 %s 失败: %v", tableName, err)
  359. }
  360. allResults = append(allResults, tableResults...)
  361. }
  362. return allResults, nil
  363. }
  364. // exportFromMultipleTablesWithPagination 从多个表导出数据(支持分页)
  365. func (r *wafLogRepository) exportFromMultipleTablesWithPagination(ctx context.Context, req adminApi.ExportWafLog, tableNames []string, page, pageSize int) ([]model.WafLogWithGatewayIP, error) {
  366. var allResults []model.WafLogWithGatewayIP
  367. currentOffset := (page - 1) * pageSize
  368. remaining := pageSize
  369. for _, tableName := range tableNames {
  370. if remaining <= 0 {
  371. break
  372. }
  373. // 先获取表的总记录数
  374. countQuery := r.buildExportQuery(ctx, req, tableName)
  375. var tableCount int64
  376. if err := countQuery.Count(&tableCount).Error; err != nil {
  377. return nil, fmt.Errorf("统计表 %s 记录数失败: %v", tableName, err)
  378. }
  379. // 如果当前偏移量大于等于表记录数,跳过这个表
  380. if currentOffset >= int(tableCount) {
  381. currentOffset -= int(tableCount)
  382. continue
  383. }
  384. // 计算在当前表中需要查询的数据
  385. tablePageSize := remaining
  386. if int(tableCount) - currentOffset < tablePageSize {
  387. tablePageSize = int(tableCount) - currentOffset
  388. }
  389. tablePage := (currentOffset / pageSize) + 1
  390. tableOffset := currentOffset % pageSize
  391. // 查询当前表数据
  392. tableResults, err := r.exportFromSingleTable(ctx, req, tableName, tablePage, tablePageSize)
  393. if err != nil {
  394. return nil, fmt.Errorf("查询表 %s 失败: %v", tableName, err)
  395. }
  396. // 如果有偏移量,跳过前面的记录
  397. if tableOffset > 0 && len(tableResults) > tableOffset {
  398. tableResults = tableResults[tableOffset:]
  399. }
  400. allResults = append(allResults, tableResults...)
  401. remaining -= len(tableResults)
  402. currentOffset = 0 // 后续表从0开始
  403. }
  404. return allResults, nil
  405. }
  406. // GetWafLogExportCount 获取导出数据总数(支持分表)
  407. func (r *wafLogRepository) GetWafLogExportCount(ctx context.Context, req adminApi.ExportWafLog) (int, error) {
  408. // 1. 解析时间范围
  409. startTime, endTime := r.getExportTimeRange(req)
  410. // 2. 获取需要查询的分表
  411. existingTables := r.Manager.GetTableNamesWithExistenceCheck(r.DBWithName(ctx, "admin"), "waf_log", startTime, endTime)
  412. if len(existingTables) == 0 {
  413. return 0, nil
  414. }
  415. var totalCount int64
  416. // 3. 逐表统计
  417. for _, tableName := range existingTables {
  418. var tableCount int64
  419. // 使用更新后的 buildExportQuery 方法
  420. query := r.buildExportQuery(ctx, req, tableName)
  421. if err := query.Count(&tableCount).Error; err != nil {
  422. return 0, fmt.Errorf("统计表 %s 记录数失败: %v", tableName, err)
  423. }
  424. totalCount += tableCount
  425. }
  426. return int(totalCount), nil
  427. }
  428. // applyWafLogFilters 应用WafLog查询过滤条件
  429. func (r *wafLogRepository) applyWafLogFilters(query *gorm.DB, req adminApi.SearchWafLogParams) *gorm.DB {
  430. if req.RequestIp != "" {
  431. trimmedName := strings.TrimSpace(req.RequestIp)
  432. query = query.Where("request_ip LIKE CONCAT('%', ?, '%')", trimmedName)
  433. }
  434. if req.Uid != 0 {
  435. query = query.Where("uid = ?", req.Uid)
  436. }
  437. if req.Api != "" {
  438. trimmedName := strings.TrimSpace(req.Api)
  439. query = query.Where("api LIKE CONCAT('%', ?, '%')", trimmedName)
  440. }
  441. if req.Name != "" {
  442. trimmedName := strings.TrimSpace(req.Name)
  443. query = query.Where("name LIKE CONCAT('%', ?, '%')", trimmedName)
  444. }
  445. if req.RuleId != 0 {
  446. query = query.Where("rule_id = ?", req.RuleId)
  447. }
  448. if req.HostId != 0 {
  449. query = query.Where("host_id = ?", req.HostId)
  450. }
  451. if req.UserAgent != "" {
  452. trimmedName := strings.TrimSpace(req.UserAgent)
  453. query = query.Where("user_agent LIKE CONCAT('%', ?, '%')", trimmedName)
  454. }
  455. if req.ApiName != "" {
  456. trimmedName := strings.TrimSpace(req.ApiName)
  457. query = query.Where("api_name LIKE CONCAT('%', ?, '%')", trimmedName)
  458. }
  459. if req.ApiType != "" {
  460. query = query.Where("api_type = ?", req.ApiType)
  461. }
  462. return query
  463. }