✨ feat(database): 完善请求日志体系并为全表补充时间字段

- 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 配置并默认启用请求日志
This commit is contained in:
chenxin committed 2026-10-09 22:08:04 +08:00
1 parent 1a4997e031
commit 126a96d744
12 files changed
+277 -55

No files matched your search

+5 -3
View File
@@ -38,9 +38,11 @@ func GenerateAudioProxyURL(originalURL, deviceID string, headers map[string]stri
if cacheDB != nil {
_, err := cacheDB.Exec(q(`
INSERT INTO audio_cache (key, original_url, device_id, headers, created_at)
VALUES (?, ?, ?, ?, ?)
`), key, originalURL, deviceID, string(headersJSON), time.Now().Unix())
INSERT INTO audio_cache (key, original_url, device_id, headers, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(key) DO UPDATE SET original_url = excluded.original_url,
device_id = excluded.device_id, headers = excluded.headers, updated_at = excluded.updated_at
`), key, originalURL, deviceID, string(headersJSON), time.Now().Unix(), time.Now().Unix())
if err != nil {
audioLogger.Error("保存音频缓存失败: " + err.Error())
}
+56 -11
View File
@@ -149,12 +149,14 @@ func createTables() error {
`CREATE TABLE IF NOT EXISTS cache (
key TEXT PRIMARY KEY,
value TEXT,
created_at INTEGER DEFAULT (strftime('%s', 'now'))
created_at INTEGER DEFAULT (strftime('%s', 'now')),
updated_at INTEGER DEFAULT (strftime('%s', 'now'))
)`,
`CREATE TABLE IF NOT EXISTS cache (
key TEXT PRIMARY KEY,
value TEXT,
created_at INTEGER DEFAULT 0
created_at BIGINT DEFAULT extract(epoch FROM now())::bigint,
updated_at BIGINT DEFAULT extract(epoch FROM now())::bigint
)`},
{"statistics",
`CREATE TABLE IF NOT EXISTS statistics (
@@ -168,6 +170,7 @@ func createTables() error {
audio_requests INTEGER DEFAULT 0,
video_requests INTEGER DEFAULT 0,
category_requests INTEGER DEFAULT 0,
created_at INTEGER DEFAULT (strftime('%s', 'now')),
updated_at INTEGER DEFAULT (strftime('%s', 'now'))
)`,
`CREATE TABLE IF NOT EXISTS statistics (
@@ -181,26 +184,29 @@ func createTables() error {
audio_requests INTEGER DEFAULT 0,
video_requests INTEGER DEFAULT 0,
category_requests INTEGER DEFAULT 0,
updated_at INTEGER DEFAULT 0
created_at BIGINT DEFAULT extract(epoch FROM now())::bigint,
updated_at BIGINT DEFAULT extract(epoch FROM now())::bigint
)`},
{"device_pool",
`CREATE TABLE IF NOT EXISTS device_pool (
id INTEGER PRIMARY KEY AUTOINCREMENT,
device_header TEXT NOT NULL,
device_header TEXT NOT NULL UNIQUE,
key TEXT NOT NULL,
key_ver INTEGER NOT NULL,
is_valid INTEGER DEFAULT 1,
created_at INTEGER NOT NULL,
last_used_at INTEGER NOT NULL
last_used_at INTEGER NOT NULL,
updated_at INTEGER DEFAULT (strftime('%s', 'now'))
)`,
`CREATE TABLE IF NOT EXISTS device_pool (
id BIGSERIAL PRIMARY KEY,
device_header TEXT NOT NULL,
device_header TEXT NOT NULL UNIQUE,
key TEXT NOT NULL,
key_ver INTEGER NOT NULL,
is_valid INTEGER DEFAULT 1,
created_at INTEGER NOT NULL,
last_used_at INTEGER NOT NULL
last_used_at INTEGER NOT NULL,
updated_at BIGINT DEFAULT extract(epoch FROM now())::bigint
)`},
{"audio_cache",
`CREATE TABLE IF NOT EXISTS audio_cache (
@@ -208,14 +214,16 @@ func createTables() error {
original_url TEXT NOT NULL,
device_id TEXT NOT NULL,
headers TEXT,
created_at INTEGER DEFAULT (strftime('%s', 'now'))
created_at INTEGER DEFAULT (strftime('%s', 'now')),
updated_at INTEGER DEFAULT (strftime('%s', 'now'))
)`,
`CREATE TABLE IF NOT EXISTS audio_cache (
key TEXT PRIMARY KEY,
original_url TEXT NOT NULL,
device_id TEXT NOT NULL,
headers TEXT,
created_at INTEGER DEFAULT 0
created_at BIGINT DEFAULT extract(epoch FROM now())::bigint,
updated_at BIGINT DEFAULT extract(epoch FROM now())::bigint
)`},
}
@@ -242,6 +250,42 @@ func createTables() error {
if _, err := cacheDB.Exec(`CREATE INDEX IF NOT EXISTS idx_device_pool_valid ON device_pool(is_valid)`); err != nil {
cacheLogger.Warning("创建device_pool索引失败: " + err.Error())
}
// 旧库补唯一约束:表已存在时 DDL 的 UNIQUE 不会生效,用唯一索引补齐;
// 若存量数据已有重复 device_header 则创建失败,仅告警提示人工清理
if _, err := cacheDB.Exec(`CREATE UNIQUE INDEX IF NOT EXISTS idx_device_pool_header ON device_pool(device_header)`); err != nil {
cacheLogger.Warning("补建device_header唯一索引失败(可能存在重复设备,清理后重启即可): " + err.Error())
}
// 旧表补 created_at/updated_at 列(新表 DDL 自带,此处仅老库生效)
for _, m := range []struct{ table, col, ddl string }{
{"cache", "updated_at", "INTEGER DEFAULT 0"}, // SQLite ALTER 不允许非常量默认值,先补 0 再回填
{"statistics", "created_at", "INTEGER DEFAULT 0"},
{"device_pool", "updated_at", "INTEGER DEFAULT 0"},
{"audio_cache", "updated_at", "INTEGER DEFAULT 0"},
} {
if dbDriver == "pgx" {
m.ddl = "BIGINT DEFAULT 0"
}
alter := "ALTER TABLE " + m.table + " ADD COLUMN "
if dbDriver == "pgx" {
alter += "IF NOT EXISTS "
}
if _, err := cacheDB.Exec(alter + m.col + " " + m.ddl); err != nil {
cacheLogger.Debug("补列跳过(已存在): " + m.table + "." + m.col)
}
}
// 旧行回填(幂等:仅 0 值行受影响)
for _, backfill := range []string{
`UPDATE cache SET updated_at = created_at WHERE updated_at = 0`,
`UPDATE statistics SET created_at = updated_at WHERE created_at = 0`,
`UPDATE device_pool SET updated_at = last_used_at WHERE updated_at = 0`,
`UPDATE audio_cache SET updated_at = created_at WHERE updated_at = 0`,
} {
if _, err := cacheDB.Exec(backfill); err != nil {
cacheLogger.Warning("时间列回填失败: " + err.Error())
}
}
return nil
}
@@ -337,8 +381,9 @@ func UpdateCache(key string, value string, expi int) error {
}
// UPSERT:ON CONFLICT 语法 SQLite 3.24+ 与 PostgreSQL 通用
_, err = cacheDB.Exec(q(`INSERT INTO cache (key, value, created_at) VALUES (?, ?, ?)
ON CONFLICT(key) DO UPDATE SET value = excluded.value`), key, string(jsonData), time.Now().Unix())
_, err = cacheDB.Exec(q(`INSERT INTO cache (key, value, created_at, updated_at) VALUES (?, ?, ?, ?)
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at`),
key, string(jsonData), time.Now().Unix(), time.Now().Unix())
if err != nil {
cacheLogger.Error("更新缓存失败: " + err.Error())
return err
+4 -2
View File
@@ -30,7 +30,8 @@ func TestConcurrentCacheAccess(t *testing.T) {
if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS cache (
key TEXT PRIMARY KEY,
value TEXT,
created_at INTEGER DEFAULT (strftime('%s', 'now'))
created_at INTEGER DEFAULT (strftime('%s', 'now')),
updated_at INTEGER DEFAULT (strftime('%s', 'now'))
)`); err != nil {
t.Fatal(err)
}
@@ -73,7 +74,8 @@ func TestWithCache(t *testing.T) {
}()
if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS cache (
key TEXT PRIMARY KEY, value TEXT,
created_at INTEGER DEFAULT (strftime('%s', 'now'))
created_at INTEGER DEFAULT (strftime('%s', 'now')),
updated_at INTEGER DEFAULT (strftime('%s', 'now'))
)`); err != nil {
t.Fatal(err)
}
+5 -5
View File
@@ -401,11 +401,11 @@ func (p *DeviceKeyPool) saveDeviceToDB(item *DeviceKeyItem) error {
// RETURNING id 同时兼容 SQLite(3.35+) 与 PostgreSQL,替代 LastInsertId()
err := cacheDB.QueryRow(q(`
INSERT INTO device_pool (device_header, key, key_ver, is_valid, created_at, last_used_at)
VALUES (?, ?, ?, ?, ?, ?)
INSERT INTO device_pool (device_header, key, key_ver, is_valid, created_at, last_used_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
RETURNING id
`), item.DeviceHeader, item.Key, item.KeyVer, boolToInt(item.IsValid),
item.CreatedAt.Unix(), item.LastUsedAt.Unix()).Scan(&item.ID)
item.CreatedAt.Unix(), item.LastUsedAt.Unix(), item.CreatedAt.Unix()).Scan(&item.ID)
return err
}
@@ -418,10 +418,10 @@ func (p *DeviceKeyPool) updateDeviceInDB(item *DeviceKeyItem) error {
_, err := cacheDB.Exec(q(`
UPDATE device_pool
SET device_header = ?, key = ?, key_ver = ?, is_valid = ?, last_used_at = ?
SET device_header = ?, key = ?, key_ver = ?, is_valid = ?, last_used_at = ?, updated_at = ?
WHERE id = ?
`), item.DeviceHeader, item.Key, item.KeyVer, boolToInt(item.IsValid),
item.LastUsedAt.Unix(), item.ID)
item.LastUsedAt.Unix(), item.LastUsedAt.Unix(), item.ID)
return err
}
+120 -23
View File
@@ -3,6 +3,7 @@ package fqapi
import (
"fanqie/internal/config"
"fmt"
"strings"
"time"
)
@@ -15,10 +16,11 @@ type RequestLogEntry struct {
IP string `json:"ip"`
Method string `json:"method"`
Path string `json:"path"`
Query string `json:"query"` // 已脱敏(key 参数替换为 ***)
Query string `json:"query"` // 解码后查询串(含 key,不脱敏)
Status int `json:"status"`
DurationMs int64 `json:"duration_ms"`
ErrorMsg string `json:"error_msg"` // 仅 status>=400 时截断保留
ErrorMsg string `json:"error_msg"` // 仅 status>=400 时截断保留
UserAgent string `json:"user_agent"` // 客户端 UA,排障时区分调用方
}
// ErrMsgMax 错误响应摘要的最大保留字节数
@@ -52,7 +54,10 @@ func InitRequestLog() error {
query TEXT,
status INTEGER,
duration_ms INTEGER,
error_msg TEXT
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 (
@@ -64,14 +69,49 @@ func InitRequestLog() error {
query TEXT,
status INTEGER,
duration_ms BIGINT,
error_msg TEXT
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)
}
if _, err := cacheDB.Exec(`CREATE INDEX IF NOT EXISTS idx_request_log_ts ON request_log(ts)`); err != nil {
cacheLogger.Warning("创建request_log索引失败: " + err.Error())
// 旧表补列(表已存在时 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)
@@ -109,7 +149,7 @@ func GetRecentRequests(limit int) []RequestLogEntry {
return nil
}
rows, err := cacheDB.Query(
q("SELECT ts, ip, method, path, query, status, duration_ms, COALESCE(error_msg, '') FROM request_log ORDER BY id DESC LIMIT ?"), limit)
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
@@ -119,7 +159,7 @@ func GetRecentRequests(limit int) []RequestLogEntry {
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); err != nil {
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
}
@@ -128,6 +168,66 @@ func GetRecentRequests(limit int) []RequestLogEntry {
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)
@@ -174,24 +274,21 @@ func reqLogWorker() {
}
}
// insertRequestLogs 事务批量写入
// insertRequestLogs 多值 INSERT 一次往返写入整批(单条语句自带原子性)
// ponytail: 批量上限 50 行 × 9 列 = 450 参数,远低于 SQLite(999)/PG(65535) 占位符上限
func insertRequestLogs(batch []RequestLogEntry) error {
tx, err := cacheDB.Begin()
if err != nil {
return err
}
stmt, err := tx.Prepare(q("INSERT INTO request_log (ts, ip, method, path, query, status, duration_ms, error_msg) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"))
if err != nil {
tx.Rollback()
return err
}
for _, e := range batch {
if _, err := stmt.Exec(e.Ts, e.IP, e.Method, e.Path, e.Query, e.Status, e.DurationMs, e.ErrorMsg); err != nil {
tx.Rollback()
return err
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)
}
return tx.Commit()
_, err := cacheDB.Exec(q(b.String()), args...)
return err
}
// cleanupRequestLog 删除超过保留期的日志