package fqapi import ( "fanqie/internal/config" "fmt" "strings" "time" ) // 请求日志:只记元数据(时间/IP/方法/路径/状态/耗时)与错误响应摘要,不存响应体。 // 全量响应与正文缓存重复且体积不可控;要看重放一个请求即可。 // RequestLogEntry 一条请求日志 type RequestLogEntry struct { Ts int64 `json:"ts"` IP string `json:"ip"` Method string `json:"method"` Path string `json:"path"` Query string `json:"query"` // 解码后查询串(含 key,不脱敏) Status int `json:"status"` DurationMs int64 `json:"duration_ms"` ErrorMsg string `json:"error_msg"` // 仅 status>=400 时截断保留 UserAgent string `json:"user_agent"` // 客户端 UA,排障时区分调用方 } // ErrMsgMax 错误响应摘要的最大保留字节数 const ErrMsgMax = 4096 const ( reqLogChanSize = 256 // channel 满则丢弃,日志绝不阻塞请求 reqLogFlushBatch = 50 // 攒批上限 reqLogFlushEvery = time.Second // 兜底 flush 间隔 reqLogCleanEvery = time.Hour // 过期清理间隔 ) var ( reqLogCh chan RequestLogEntry reqLogQuit chan struct{} reqLogDone chan struct{} ) // InitRequestLog 建表并启动后台落库协程(未启用时为空操作) func InitRequestLog() error { if !requestLogEnabled() { return nil } ddl := `CREATE TABLE IF NOT EXISTS request_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL, ip TEXT, method TEXT, path TEXT, query TEXT, status INTEGER, duration_ms INTEGER, error_msg TEXT, user_agent TEXT, created_at INTEGER DEFAULT (strftime('%s', 'now')), updated_at INTEGER DEFAULT (strftime('%s', 'now')) )` if dbDriver == "pgx" { ddl = `CREATE TABLE IF NOT EXISTS request_log ( id BIGSERIAL PRIMARY KEY, ts BIGINT NOT NULL, ip TEXT, method TEXT, path TEXT, query TEXT, status INTEGER, duration_ms BIGINT, error_msg TEXT, user_agent TEXT, created_at BIGINT DEFAULT extract(epoch FROM now())::bigint, updated_at BIGINT DEFAULT extract(epoch FROM now())::bigint )` } if _, err := cacheDB.Exec(ddl); err != nil { return fmt.Errorf("创建request_log表失败: %w", err) } // 旧表补列(表已存在时 CREATE 不会加列;PG 支持 IF NOT EXISTS,SQLite 报错即视为已存在) if dbDriver == "pgx" { for _, col := range []string{ `ALTER TABLE request_log ADD COLUMN IF NOT EXISTS user_agent TEXT`, `ALTER TABLE request_log ADD COLUMN IF NOT EXISTS created_at BIGINT DEFAULT 0`, `ALTER TABLE request_log ADD COLUMN IF NOT EXISTS updated_at BIGINT DEFAULT 0`, } { if _, err := cacheDB.Exec(col); err != nil { cacheLogger.Warning("补加列失败: " + err.Error()) } } } else { for _, col := range []string{ `ALTER TABLE request_log ADD COLUMN user_agent TEXT`, `ALTER TABLE request_log ADD COLUMN created_at INTEGER DEFAULT 0`, `ALTER TABLE request_log ADD COLUMN updated_at INTEGER DEFAULT 0`, } { if _, err := cacheDB.Exec(col); err != nil { cacheLogger.Debug("补列跳过(已存在): " + err.Error()) } } } // append-only 日志表:旧行按 ts 回填两个时间列(幂等) if _, err := cacheDB.Exec(q(`UPDATE request_log SET created_at = ts, updated_at = ts WHERE created_at = 0`)); err != nil { cacheLogger.Warning("request_log时间列回填失败: " + err.Error()) } for _, idx := range []string{ `CREATE INDEX IF NOT EXISTS idx_request_log_ts ON request_log(ts)`, `CREATE INDEX IF NOT EXISTS idx_request_log_ip ON request_log(ip)`, `CREATE INDEX IF NOT EXISTS idx_request_log_status ON request_log(status)`, } { if _, err := cacheDB.Exec(idx); err != nil { cacheLogger.Warning("创建request_log索引失败: " + err.Error()) } } reqLogCh = make(chan RequestLogEntry, reqLogChanSize) reqLogQuit = make(chan struct{}) reqLogDone = make(chan struct{}) go reqLogWorker() cacheLogger.Info("请求日志已启用") return nil } // CloseRequestLog 停止后台协程并落盘剩余日志 func CloseRequestLog() { if reqLogQuit == nil { return } close(reqLogQuit) <-reqLogDone reqLogQuit = nil } // LogRequest 非阻塞投递一条请求日志(未启用/队列满时静默丢弃) func LogRequest(e RequestLogEntry) { if reqLogCh == nil { return } select { case reqLogCh <- e: default: } } // GetRecentRequests 查询最近 limit 条请求日志 func GetRecentRequests(limit int) []RequestLogEntry { if cacheDB == nil || reqLogCh == nil { return nil } rows, err := cacheDB.Query( q("SELECT ts, ip, method, path, query, status, duration_ms, COALESCE(error_msg, ''), COALESCE(user_agent, '') FROM request_log ORDER BY id DESC LIMIT ?"), limit) if err != nil { cacheLogger.Warning("查询请求日志失败: " + err.Error()) return nil } defer rows.Close() entries := make([]RequestLogEntry, 0, limit) for rows.Next() { var e RequestLogEntry if err := rows.Scan(&e.Ts, &e.IP, &e.Method, &e.Path, &e.Query, &e.Status, &e.DurationMs, &e.ErrorMsg, &e.UserAgent); err != nil { cacheLogger.Warning("读取请求日志失败: " + err.Error()) break } entries = append(entries, e) } return entries } // DailyStat 一天的请求量统计 type DailyStat struct { Day string `json:"day"` // 本地时区 MM-DD Count int64 `json:"count"` Pct int `json:"pct"` // 相对窗口峰值的百分比(进度条宽度) MaxCount bool `json:"max_count"` // 是否为窗口内峰值(前端高亮用) } // GetDailyRequestStats 按本地时区分天统计最近 days 天(含今天)的请求量,依赖 request_log。 // ponytail: (ts+固定偏移)/86400 分组,DST 切换日边界误差 1 小时,趋势展示可接受 func GetDailyRequestStats(days int) []DailyStat { if cacheDB == nil || reqLogCh == nil || days <= 0 { return nil } _, off := time.Now().Zone() day := func(ts int64) int64 { return (ts + int64(off)) / 86400 } cutoff := (day(time.Now().Unix()) - int64(days) + 1) * 86400 - int64(off) rows, err := cacheDB.Query(q( "SELECT (ts + ?) / 86400, COUNT(*) FROM request_log WHERE ts >= ? GROUP BY 1 ORDER BY 1"), int64(off), cutoff) if err != nil { cacheLogger.Warning("按天聚合请求量失败: " + err.Error()) return nil } defer rows.Close() counts := make(map[int64]int64, days) var maxCount int64 for rows.Next() { var d, c int64 if err := rows.Scan(&d, &c); err != nil { cacheLogger.Warning("读取按天统计失败: " + err.Error()) break } counts[d] = c if c > maxCount { maxCount = c } } // 补齐无请求的日期,升序返回 stats := make([]DailyStat, 0, days) for i := int64(days) - 1; i >= 0; i-- { d := day(time.Now().Unix()) - i c := counts[d] var pct int if maxCount > 0 { pct = int(c * 100 / maxCount) } stats = append(stats, DailyStat{ Day: time.Unix(d*86400-int64(off), 0).Format("01-02"), Count: c, Pct: pct, MaxCount: c > 0 && c == maxCount, }) } return stats } // reqLogWorker 攒批落库 + 定期清理过期日志 func reqLogWorker() { defer close(reqLogDone) batch := make([]RequestLogEntry, 0, reqLogFlushBatch) flush := func() { if len(batch) == 0 { return } if err := insertRequestLogs(batch); err != nil { cacheLogger.Warning("写入请求日志失败: " + err.Error()) } batch = batch[:0] } defer flush() ticker := time.NewTicker(reqLogFlushEvery) cleanTicker := time.NewTicker(reqLogCleanEvery) defer ticker.Stop() defer cleanTicker.Stop() for { select { case e := <-reqLogCh: batch = append(batch, e) if len(batch) >= reqLogFlushBatch { flush() } case <-ticker.C: flush() case <-cleanTicker.C: cleanupRequestLog() case <-reqLogQuit: // 排空 channel 后退出 for { select { case e := <-reqLogCh: batch = append(batch, e) default: return } } } } } // insertRequestLogs 多值 INSERT 一次往返写入整批(单条语句自带原子性) // ponytail: 批量上限 50 行 × 9 列 = 450 参数,远低于 SQLite(999)/PG(65535) 占位符上限 func insertRequestLogs(batch []RequestLogEntry) error { var b strings.Builder b.WriteString("INSERT INTO request_log (ts, ip, method, path, query, status, duration_ms, error_msg, user_agent) VALUES ") args := make([]interface{}, 0, len(batch)*9) for i, e := range batch { if i > 0 { b.WriteByte(',') } b.WriteString("(?, ?, ?, ?, ?, ?, ?, ?, ?)") args = append(args, e.Ts, e.IP, e.Method, e.Path, e.Query, e.Status, e.DurationMs, e.ErrorMsg, e.UserAgent) } _, err := cacheDB.Exec(q(b.String()), args...) return err } // cleanupRequestLog 删除超过保留期的日志 func cleanupRequestLog() { if !requestLogEnabled() { return } days := 7 if cfg := config.GetAppConfig(); cfg != nil && cfg.Server.RequestLog.RetentionDays > 0 { days = cfg.Server.RequestLog.RetentionDays } cutoff := time.Now().Unix() - int64(days)*86400 if result, err := cacheDB.Exec(q("DELETE FROM request_log WHERE ts < ?"), cutoff); err != nil { cacheLogger.Warning("清理请求日志失败: " + err.Error()) } else if n, _ := result.RowsAffected(); n > 0 { cacheLogger.Info(fmt.Sprintf("已清理 %d 条过期请求日志(保留 %d 天)", n, days)) } } // requestLogEnabled 配置启用且数据库可用 func requestLogEnabled() bool { cfg := config.GetAppConfig() return cfg != nil && cfg.Server.RequestLog.Enabled && cacheDB != nil }