From e8ef73321c791cbfa44f71556cb84757481e7485 Mon Sep 17 00:00:00 2001 From: JianFeeeee <109188060+JianFeeeee@users.noreply.github.com> Date: Tue, 11 Aug 2026 08:42:13 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20restart=20wipes=20stats=20history=20?= =?UTF-8?q?=E2=80=94=20LoadAudit=20replayed=20only=20the=20last=203000=20l?= =?UTF-8?q?ines=20(92%=20of=20which=20are=203s=20access=20events)=20and=20?= =?UTF-8?q?fed=20access=20rows=20into=20aggregates;=20now=20replays=20ever?= =?UTF-8?q?y=20real=20request=20into=20totals/quota=20windows,=20skips=20e?= =?UTF-8?q?vent=20rows,=20tolerates=20>64KB=20lines;=20+TestLoadAuditFullR?= =?UTF-8?q?eplay?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/gateway/stats.go | 50 ++++++++++++++++++++++------------ internal/gateway/stats_test.go | 47 ++++++++++++++++++++++++++++++++ plan.md | 5 ++-- 3 files changed, 82 insertions(+), 20 deletions(-) diff --git a/internal/gateway/stats.go b/internal/gateway/stats.go index 5eb329b..ef338eb 100644 --- a/internal/gateway/stats.go +++ b/internal/gateway/stats.go @@ -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 ..old once it diff --git a/internal/gateway/stats_test.go b/internal/gateway/stats_test.go index 84f077d..cb4c238 100644 --- a/internal/gateway/stats_test.go +++ b/internal/gateway/stats_test.go @@ -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) + } } \ No newline at end of file diff --git a/plan.md b/plan.md index b30190d..b833883 100644 --- a/plan.md +++ b/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 ..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 → 观察。 \ No newline at end of file +- [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/超大行混合回放断言）。 \ No newline at end of file