mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-09-20 00:48:00 +00:00
fix: restart wipes stats history — LoadAudit replayed only the last 3000 lines (92% of which are 3s access events) and fed access rows into aggregates; now replays every real request into totals/quota windows, skips event rows, tolerates >64KB lines; +TestLoadAuditFullReplay
This commit is contained in:
@ -135,21 +135,30 @@ func incStatus(a *Stat, name string, r Req) {
|
||||
func (s *Stats) LoadAudit(path string) {
|
||||
f, err := os.Open(path)
|
||||
if err == nil {
|
||||
defer f.Close()
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
var recs []Req
|
||||
// tolerate long error summaries / oversized junk lines
|
||||
sc := bufio.NewScanner(f)
|
||||
sc.Buffer(make([]byte, 64*1024), 16*1024*1024)
|
||||
for sc.Scan() {
|
||||
var r Req
|
||||
if json.Unmarshal(sc.Bytes(), &r) == nil {
|
||||
recs = append(recs, r)
|
||||
if json.Unmarshal(sc.Bytes(), &r) != nil || r.Type == "" {
|
||||
continue // access/event rows (obj) and malformed lines are no requests
|
||||
}
|
||||
// replay EVERY request row into the aggregates so totals and
|
||||
// quota windows survive restarts; only the ring buffer view stays
|
||||
// capped at maxRecs.
|
||||
s.aggregateLocked(r)
|
||||
recs = append(recs, r)
|
||||
}
|
||||
_ = f.Close()
|
||||
if len(recs) > s.maxRecs {
|
||||
recs = recs[len(recs)-s.maxRecs:]
|
||||
}
|
||||
for _, r := range recs {
|
||||
s.Record(r)
|
||||
if n := len(recs); n > s.maxRecs {
|
||||
recs = append([]Req(nil), recs[n-s.maxRecs:]...)
|
||||
}
|
||||
s.recs = recs
|
||||
s.auditPath = path
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
s.auditPath = path
|
||||
@ -160,6 +169,20 @@ func (s *Stats) LoadAudit(path string) {
|
||||
func (s *Stats) Record(r Req) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.aggregateLocked(r)
|
||||
s.recs = append(s.recs, r)
|
||||
if len(s.recs) > s.maxRecs {
|
||||
s.recs = s.recs[len(s.recs)-s.maxRecs:]
|
||||
}
|
||||
if s.auditPath != "" {
|
||||
s.rotateAuditLocked()
|
||||
appendAuditLine(s.auditPath, r)
|
||||
}
|
||||
}
|
||||
|
||||
// aggregateLocked folds r into every aggregate row and the quota window
|
||||
// bucket. Caller must hold s.mu.
|
||||
func (s *Stats) aggregateLocked(r Req) {
|
||||
inc(s.byKey, r.Key, r)
|
||||
if r.Model != "" {
|
||||
inc(s.byModel, r.Model, r)
|
||||
@ -180,13 +203,12 @@ func (s *Stats) Record(r Req) {
|
||||
}
|
||||
inc(ks, r.Source, r)
|
||||
if r.Status > 0 {
|
||||
name := strconv.Itoa(r.Status)
|
||||
a := s.byStatus[r.Status]
|
||||
if a == nil {
|
||||
a = &Stat{}
|
||||
s.byStatus[r.Status] = a
|
||||
}
|
||||
incStatus(a, name, r)
|
||||
incStatus(a, strconv.Itoa(r.Status), r)
|
||||
}
|
||||
// window bucket for quota enforcement (per source-model pair, per unix hour)
|
||||
tok := r.Prompt + r.Compl
|
||||
@ -210,14 +232,6 @@ func (s *Stats) Record(r Req) {
|
||||
}
|
||||
}
|
||||
}
|
||||
s.recs = append(s.recs, r)
|
||||
if len(s.recs) > s.maxRecs {
|
||||
s.recs = s.recs[len(s.recs)-s.maxRecs:]
|
||||
}
|
||||
if s.auditPath != "" {
|
||||
s.rotateAuditLocked()
|
||||
appendAuditLine(s.auditPath, r)
|
||||
}
|
||||
}
|
||||
|
||||
// rotateAuditLocked renames the audit file to <path>.<unix>.old once it
|
||||
|
||||
@ -4,6 +4,7 @@ import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
@ -85,4 +86,50 @@ func TestAuditRotationRecords(t *testing.T) {
|
||||
if len(matches) != 1 {
|
||||
t.Fatalf("Record must rotate too: got %d old files", len(matches))
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadAuditFullReplay(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "audit.jsonl")
|
||||
lines := []string{
|
||||
`{"obj":"access","time":1699999999000,"key":"k","method":"GET","path":"/api/stats","status":200}`,
|
||||
`{"time":1700000000000,"key":"k","type":"chat","model":"m","source":"s","prompt_tokens":100,"completion_tokens":50,"latency_ms":10,"ok":true,"status":200}`,
|
||||
`{this is not valid json`,
|
||||
`{"time":1700003600000,"key":"k","type":"stream","model":"m","source":"s","prompt_tokens":200,"completion_tokens":20,"latency_ms":20,"ok":false,"status":503}`,
|
||||
"garbage-not-json\n",
|
||||
`{"time":1700007200000,"key":"k","type":"chat","model":"m2","source":"s","prompt_tokens":7,"completion_tokens":3,"latency_ms":5,"ok":true,"status":200}`,
|
||||
}
|
||||
loaded := strings.Join(lines, "\n") + "\n" + strings.Repeat("x", 1<<18) + "\n"
|
||||
// oversized row at the END proves the scanner tolerates >64KB lines and
|
||||
// still finishes the replay instead of truncating silently.
|
||||
loaded += `{"time":1700010800000,"key":"k","type":"chat","model":"m","source":"s","prompt_tokens":1,"completion_tokens":1,"latency_ms":1,"ok":true,"status":200}` + "\n"
|
||||
if err := os.WriteFile(path, []byte(loaded), 0644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
s := NewStats(1000)
|
||||
s.LoadAudit(path)
|
||||
|
||||
got := s.byModel["m"]
|
||||
if got == nil || got.Tokens != 100+50+200+20+1+1 {
|
||||
t.Fatalf("aggregates must replay EVERY request row, got %#v", got)
|
||||
}
|
||||
if s.byModel["m2"] == nil || s.byModel["m2"].Tokens != 10 {
|
||||
t.Fatalf("m2 must be replayed too, got %#v", s.byModel["m2"])
|
||||
}
|
||||
// access rows are not requests: 4 real rows, junk skipped
|
||||
if len(s.recs) != 4 {
|
||||
t.Fatalf("ring must hold only real requests, got %d rows: %#v", len(s.recs), s.recs)
|
||||
}
|
||||
// quota window rebuilt from full history
|
||||
if w := s.WindowTokens("m", "s", hourSec); w != 372 {
|
||||
t.Fatalf("window tokens want 372, got %d", w)
|
||||
}
|
||||
// by_status only from requests (200 x3, 503 x1) — access line must not count
|
||||
if st := s.byStatus[200]; st == nil || st.Reqs != 3 {
|
||||
t.Fatalf("by_status 200 want 3 reqs, got %#v", st)
|
||||
}
|
||||
if s.byStatus[200].Err != 0 || s.byStatus[503] == nil || s.byStatus[503].Reqs != 1 {
|
||||
t.Fatalf("by_status wrong: %#v", s.byStatus)
|
||||
}
|
||||
}
|
||||
5
plan.md
5
plan.md
@ -224,5 +224,6 @@ p==nil → 400/404;!TryAcquire → 429 busy(快速失败)
|
||||
- [x] **P3-3(2026-08-10)审计分类与 jsonl 轮转**:`Stats.byStatus map[int]*Stat`(402 欠费/400 schema 等按状态码独立计数,不触发 provider 退避,`Snapshot.by_status` 有序输出);audit 文件超 `auditRotateBytes`(64MB, var 可测) 轮转 `rename <path>.<unix>.old` 并保留最新 `auditKeepOld`(10) 份——`rotateAuditLocked` 持 mu 在 `Record`/`AppendAudit` 内触发。
|
||||
- [x] **P3-4(2026-08-10)运维项(代码部分)**:`main.go` 启动告警——gateway_keys 为空 / 命中种子 key(sk-gw-local-0001 等)提示轮换、listen 绑定 0.0.0.0/:: 提示收敛内网。生产实践(换 admin key、内网绑定)随本次上线执行。
|
||||
- [x] **P3-5(2026-08-10)测试**:新增 `stats_test.go`(by_status 断言、轮转→保留上限→Record 路径轮转);gateway 新增 `TestAutoStatesReportChainHealth`(states 契约:失败后 fail_count>0+cooling,PUT 复位归零);`go vet ./...` + `go test ./...` 全绿。
|
||||
- [x] **P3-6(2026-08-10)WebUI 右键菜单无法关闭(用户实测)**:根因——关闭依赖 `document` 冒泡阶段 once-click 监听,而优先级页块/密钥砖自己的 click 处理 `stopPropagation()` 阻断冒泡 → 菜单永不关闭;修复:改为 **document 捕获阶段**全局 `click`+`mousedown` 关闭(捕获先于一切目标处理器,stopPropagation 无法拦截),优先级页与密钥页共用 `showCtx` 一并修复。
|
||||
- [ ] **P3-7**:推送 origin → 生产机 pull → `-tags luajit` 全量回归(含 P2-7)→ 部署 `/usr/local/bin/llmsproxy` + 重启 service → 观察。
|
||||
- [x] **P3-6(2026-08-10)WebUI 右键菜单无法关闭(用户实测)**:根因——`showCtx` 创建菜单后从未赋值 `ctxEl`(`hideCtx()` 恒为空操作),任何路径(点菜单项/点外部/二次右键)都关不掉;补 `ctxEl = w`,优先级页与密钥页共用 `showCtx` 一并修复(这也是最初版本就存在的 bug)。
|
||||
- [ ] **P3-7**:推送 origin → 生产机 pull → `-tags luajit` 全量回归(含 P2-7)→ 部署 `/usr/local/bin/llmsproxy` + 重启 service → 观察。
|
||||
- [x] **P3-8(2026-08-11)审计回放修复(用户实测)**:重启后统计只剩最后 3000 行≈5.3M tokens,历史 273M 消失、且记录里全是 access 脏行——根因 `LoadAudit`:① 只取文件尾 3000 行(而 92% 行是每 3s 的 access 事件);② access/事件行(type 空)被当请求灌入聚合;③ Scanner 默认 64KB 截断风险。修复:回放**全部真实请求行**进聚合(恢复 totals/配额窗口),仅视图环形缓冲限 maxRecs;跳过 type 空的行;`sc.Buffer` 抬到 16MB。新增 `TestLoadAuditFullReplay`(access/坏 JSON/超大行混合回放断言)。
|
||||
Reference in New Issue
Block a user