chenxin
e3d96ca637
- request_log 新增 user_agent 列（含旧表 ALTER 迁移）及 ip/status 索引，记录解码后的明文 query（含 key，不脱敏） - 请求日志批量写入改为多值 INSERT 单次往返，替代事务逐条 Exec - 新增按本地时区分天聚合查询，状态页展示今日请求量与近 7 天请求趋势（峰值日高亮） - 新增 TZ 时区支持：.env 配置 TZ 后日志时间戳与状态页全部时间统一生效，二进制内嵌 tzdata 保证容器可用 - 五张表统一补齐 created_at/updated_at 字段，启动时自动补列并按业务时间回填旧行 - device_pool 的 device_header 增加唯一约束；audio_cache 写入改为 UPSERT 修复同 key 重复报错 - 状态页"最近请求"新增 UA 列与限流状态显示；config.yaml 暴露 rate_limit_qps 配置并默认启用请求日志 - deploy.sh 同名版本默认拒绝覆盖，新增 --force/-f 参数显式允许覆盖
316 lines
9.3 KiB
Go
316 lines
9.3 KiB
Go
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
|
||
}
|