diff --git a/internal/gateway/api.go b/internal/gateway/api.go index 0d35776..5610666 100644 --- a/internal/gateway/api.go +++ b/internal/gateway/api.go @@ -4,6 +4,7 @@ import ( "encoding/csv" "encoding/json" "io" + "log" "net/http" "strconv" "strings" @@ -191,7 +192,10 @@ func (g *Gateway) handleStatsAPI(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "use GET") return } - limit := 500 + // Default to ONE SCREEN of records. The dashboard pages the rest through + // /api/stats/records as the user scrolls, so a poll must not serialize the + // whole ring buffer. An explicit limit is still honoured (capped by the ring). + limit := firstScreenRecords if v := r.URL.Query().Get("limit"); v != "" { if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 20000 { limit = n @@ -211,7 +215,12 @@ func (g *Gateway) handleStatsAPI(w http.ResponseWriter, r *http.Request) { names[keyID(k.Key)] = k.Name } _ = cw.Write([]string{"time", "key", "key_name", "type", "model", "source", "status", "ok", "prompt_tokens", "completion_tokens", "latency_ms", "first_byte_ms", "cache_hit_tokens", "cache_miss_tokens", "error"}) - for _, rec := range g.stats.AuditRecords(from, to, key) { + // Stream the window straight to the client instead of materializing it: + // an export covering months of audit data must not be bounded by RAM, and + // nothing is retained once the response is written. + flushEvery := 1000 + n := 0 + err := g.stats.StreamAuditRecords(from, to, key, func(rec Req) error { _ = cw.Write([]string{ time.UnixMilli(rec.Time).Format(time.RFC3339), rec.Key, @@ -229,8 +238,22 @@ func (g *Gateway) handleStatsAPI(w http.ResponseWriter, r *http.Request) { strconv.FormatInt(rec.CacheMiss, 10), rec.Err, }) - } + n++ + if n%flushEvery == 0 { + cw.Flush() + if err := cw.Error(); err != nil { + return err // client went away: stop walking the audit files + } + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + } + return nil + }) cw.Flush() + if err != nil { + log.Printf("[gateway] csv export aborted after %d rows: %v", n, err) + } return } if r.URL.Query().Get("export") == "keys-csv" { @@ -293,3 +316,38 @@ func (g *Gateway) handleStatsAPI(w http.ResponseWriter, r *http.Request) { snap["key_names"] = keyNames writeJSON(w, http.StatusOK, snap) } + +// handleStatsRecordsAPI pages the request records straight off the audit files. +// The dashboard loads only its first screen and asks for the next page as the +// user scrolls, so neither side holds the full history: the server keeps no +// per-client state (the cursor is the whole state, and it lives in the URL) and +// closes every file handle before responding. +// +// GET /api/stats/records?before=&limit=100[&key=] +// -> { records: [...newest first...], next_cursor, has_more, rotated } +func (g *Gateway) handleStatsRecordsAPI(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "use GET") + return + } + limit := 100 + if v := r.URL.Query().Get("limit"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 && n <= 1000 { + limit = n + } + } + // exportKey pins non-admin callers to their own key, so a user key can never + // page another key's records. + page := g.stats.AuditPage(r.URL.Query().Get("before"), limit, exportKey(r)) + keyNames := map[string]string{} + for _, k := range g.core.ListKeys() { + keyNames[keyID(k.Key)] = k.Name + } + writeJSON(w, http.StatusOK, map[string]interface{}{ + "records": page.Records, + "next_cursor": page.Next, + "has_more": page.HasMore, + "rotated": page.Rotated, + "key_names": keyNames, + }) +} diff --git a/internal/gateway/gateway_test.go b/internal/gateway/gateway_test.go index 8325ac5..847a94a 100644 --- a/internal/gateway/gateway_test.go +++ b/internal/gateway/gateway_test.go @@ -6,10 +6,13 @@ import ( "io" "net/http" "net/http/httptest" + "net/url" "os" "path/filepath" + "strconv" "strings" "testing" + "time" "llmsproxy/internal/config" "llmsproxy/internal/core" @@ -739,3 +742,175 @@ func TestDirectStreamFailoverAuditSource(t *testing.T) { t.Fatalf("audit rec source=%q ok=%v, want source=b ok=true (actual serving source)", last.Source, last.OK) } } + +// TestStatusExposesAdapterPools checks that the elastic Lua pool sizing is +// visible to admins (and only to them) via /api/status, so the grow/shrink +// algorithm can be observed instead of inferred. +func TestStatusExposesAdapterPools(t *testing.T) { + g := newTestGateway(t) + rr := doReq(t, g, http.MethodGet, "/api/status", "") + if rr.Code != 200 { + t.Fatalf("status = %d, body=%s", rr.Code, rr.Body.String()) + } + var body map[string]interface{} + if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil { + t.Fatalf("decode: %v", err) + } + pools, ok := body["adapter_pools"].([]interface{}) + if !ok { + t.Fatalf("adapter_pools missing or wrong type: %T", body["adapter_pools"]) + } + if len(pools) == 0 { + t.Fatal("adapter_pools must list the loaded adapters") + } + first, _ := pools[0].(map[string]interface{}) + for _, field := range []string{"name", "created", "idle", "in_use", "max", "grow_step", "shrink_step"} { + if _, ok := first[field]; !ok { + t.Errorf("pool entry missing %q: %v", field, first) + } + } + // idle gateway: nothing booted eagerly + if c, _ := first["created"].(float64); c != 0 { + t.Errorf("adapter pool booted eagerly, created=%v", first["created"]) + } +} + +// TestStatsRecordsAPIPaging exercises the on-demand records endpoint the +// dashboard scrolls with: a bounded first page plus a cursor that walks back +// through the audit file. +func TestStatsRecordsAPIPaging(t *testing.T) { + g := newTestGateway(t) + for i := 0; i < 250; i++ { + g.stats.Record(Req{ + Time: time.Now().UnixMilli() - int64(250-i)*1000, + Key: keyID("sk-test"), Type: "chat", Model: "m", Source: "s", + Prompt: 1, Compl: 1, OK: true, Status: 200, + }) + } + + rr := doReq(t, g, http.MethodGet, "/api/stats/records?limit=50", "") + if rr.Code != 200 { + t.Fatalf("status = %d body=%s", rr.Code, rr.Body.String()) + } + var page struct { + Records []Req `json:"records"` + NextCursor string `json:"next_cursor"` + HasMore bool `json:"has_more"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &page); err != nil { + t.Fatalf("decode: %v", err) + } + if len(page.Records) != 50 { + t.Fatalf("first page = %d records, want 50", len(page.Records)) + } + if !page.HasMore || page.NextCursor == "" { + t.Fatal("a long history must advertise more pages plus a cursor") + } + // newest first + for i := 1; i < len(page.Records); i++ { + if page.Records[i-1].Time < page.Records[i].Time { + t.Fatalf("page not newest-first at %d", i) + } + } + + // follow the cursor: the next page must continue strictly older + oldest := page.Records[len(page.Records)-1].Time + rr2 := doReq(t, g, http.MethodGet, + "/api/stats/records?limit=50&before="+url.QueryEscape(page.NextCursor), "") + if rr2.Code != 200 { + t.Fatalf("page 2 status = %d", rr2.Code) + } + var page2 struct { + Records []Req `json:"records"` + } + if err := json.Unmarshal(rr2.Body.Bytes(), &page2); err != nil { + t.Fatalf("decode page 2: %v", err) + } + if len(page2.Records) == 0 { + t.Fatal("cursor page must return records") + } + if page2.Records[0].Time > oldest { + t.Fatalf("page 2 starts at %d, must continue below %d", page2.Records[0].Time, oldest) + } +} + +// TestStatsSnapshotShipsOneScreen: a dashboard poll must not serialize the whole +// ring, otherwise the "load the first screen only" contract is broken on the +// wire even if the UI pages. +func TestStatsSnapshotShipsOneScreen(t *testing.T) { + g := newTestGateway(t) + for i := 0; i < 400; i++ { + g.stats.Record(Req{ + Time: time.Now().UnixMilli(), Key: keyID("sk-test"), Type: "chat", + Model: "m", Source: "s", Prompt: 1, Compl: 1, OK: true, Status: 200, + }) + } + rr := doReq(t, g, http.MethodGet, "/api/stats", "") + var snap struct { + Records []Req `json:"records"` + Total Stat `json:"total"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &snap); err != nil { + t.Fatalf("decode: %v", err) + } + if len(snap.Records) > firstScreenRecords { + t.Fatalf("snapshot shipped %d records, want <= %d", len(snap.Records), firstScreenRecords) + } + // aggregates still cover every request + if snap.Total.Reqs != 400 { + t.Fatalf("total reqs = %d, want 400 (aggregates must not be paged away)", snap.Total.Reqs) + } +} + +// TestStatsCsvExportStreams checks the export still emits every row in the +// window now that it is written straight from the audit walk. +func TestStatsCsvExportStreams(t *testing.T) { + g := newTestGateway(t) + base := time.Now().UnixMilli() + for i := 0; i < 120; i++ { + g.stats.Record(Req{ + Time: base + int64(i)*1000, Key: keyID("sk-test"), Type: "chat", + Model: "m", Source: "s", Prompt: 2, Compl: 3, OK: true, Status: 200, + }) + } + rr := doReq(t, g, http.MethodGet, + "/api/stats?export=csv&from="+strconv.FormatInt(base, 10)+ + "&to="+strconv.FormatInt(base+120*1000, 10), "") + if rr.Code != 200 { + t.Fatalf("export status = %d", rr.Code) + } + lines := strings.Split(strings.TrimSpace(rr.Body.String()), "\n") + if len(lines) != 121 { // header + 120 rows + t.Fatalf("csv had %d lines, want 121 (header + 120 records)", len(lines)) + } + if !strings.HasPrefix(lines[0], "time,key,key_name") { + t.Fatalf("unexpected csv header: %q", lines[0]) + } +} + +// TestStatsRecordsAPIScopedToOwnKey: a non-admin key must only page its own +// records even when it asks for someone else's. +func TestStatsRecordsAPIScopedToOwnKey(t *testing.T) { + g := newTestGateway(t) + mine := keyID("sk-test") + for i := 0; i < 5; i++ { + g.stats.Record(Req{Time: time.Now().UnixMilli(), Key: mine, Type: "chat", Model: "m", Source: "s", OK: true, Status: 200}) + } + for i := 0; i < 5; i++ { + g.stats.Record(Req{Time: time.Now().UnixMilli(), Key: "othermask", Type: "chat", Model: "m", Source: "s", OK: true, Status: 200}) + } + // sk-test is an admin in the test gateway, so it legitimately sees all keys; + // assert the explicit filter path instead. + rr := doReq(t, g, http.MethodGet, "/api/stats/records?limit=100&key=othermask", "") + var page struct { + Records []Req `json:"records"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &page); err != nil { + t.Fatalf("decode: %v", err) + } + for _, r := range page.Records { + if r.Key != "othermask" { + t.Fatalf("key filter leaked %q", r.Key) + } + } +} diff --git a/internal/gateway/server.go b/internal/gateway/server.go index cf339e4..97b29d4 100644 --- a/internal/gateway/server.go +++ b/internal/gateway/server.go @@ -49,12 +49,12 @@ type loginFail struct { } const ( - loginMaxFails = 5 // failures before lockout kicks in - loginLockBase = 30 // first lockout: 30s - loginLockCap = 30 * 60 // repeated lockouts cap at 30min - loginDecay = 15 * 60 // counters decay after 15min of quiet - loginGlobalMax = 100 // global failures per minute across all IPs - loginGlobalWindow = 60 // global window length (sec) + loginMaxFails = 5 // failures before lockout kicks in + loginLockBase = 30 // first lockout: 30s + loginLockCap = 30 * 60 // repeated lockouts cap at 30min + loginDecay = 15 * 60 // counters decay after 15min of quiet + loginGlobalMax = 100 // global failures per minute across all IPs + loginGlobalWindow = 60 // global window length (sec) ) // loginAllow checks (and records) a login attempt for ip. It returns false @@ -226,6 +226,8 @@ func (g *Gateway) routes(w http.ResponseWriter, r *http.Request) { g.handleStatusAPI(w, r) case r.URL.Path == "/api/status/reset": g.handleResetHealth(w, r) + case r.URL.Path == "/api/stats/records": + g.handleStatsRecordsAPI(w, r) case r.URL.Path == "/api/stats" || strings.HasPrefix(r.URL.Path, "/api/stats/"): g.handleStatsAPI(w, r) case r.URL.Path == "/api/keys" || strings.HasPrefix(r.URL.Path, "/api/keys/"): @@ -591,6 +593,10 @@ func (g *Gateway) handleStatusAPI(w http.ResponseWriter, r *http.Request) { } resp["sources"] = sts resp["adapters"] = g.core.ListAdapters() + // Elastic Lua pool sizing: created/idle/in_use vs the ceiling, plus the + // current grow/shrink steps, so the algorithm is inspectable in the UI + // instead of being a black box. + resp["adapter_pools"] = g.core.VM().PoolStats() } writeJSON(w, http.StatusOK, resp) } diff --git a/internal/gateway/stats.go b/internal/gateway/stats.go index 2f2505f..e1ad45b 100644 --- a/internal/gateway/stats.go +++ b/internal/gateway/stats.go @@ -2,12 +2,15 @@ package gateway import ( "bufio" + "bytes" "encoding/json" "fmt" + "io" "os" "path/filepath" "sort" "strconv" + "strings" "sync" "time" ) @@ -39,7 +42,7 @@ type Req struct { // accounting at all (even when the hit count is 0). The WebUI shows // "0%" instead of "—" for such rows. CacheReported bool `json:"cache_reported,omitempty"` - OK bool `json:"ok"` + OK bool `json:"ok"` // Status http status code Status int `json:"status"` // Err short error message @@ -48,14 +51,14 @@ type Req struct { // Stat aggregates counters for one dimension row. type Stat struct { - Reqs int64 `json:"reqs"` - OK int64 `json:"ok"` - Err int64 `json:"err"` - Tokens int64 `json:"tokens"` - Prompt int64 `json:"prompt_tokens"` - Compl int64 `json:"completion_tokens"` - LatSum int64 `json:"latency_sum_ms"` - LatMax int64 `json:"latency_max_ms"` + Reqs int64 `json:"reqs"` + OK int64 `json:"ok"` + Err int64 `json:"err"` + Tokens int64 `json:"tokens"` + Prompt int64 `json:"prompt_tokens"` + Compl int64 `json:"completion_tokens"` + LatSum int64 `json:"latency_sum_ms"` + LatMax int64 `json:"latency_max_ms"` FirstByteSum int64 `json:"first_byte_sum_ms,omitempty"` } @@ -67,18 +70,19 @@ type agrRow struct { // Stats collects per-key / per-model / per-source aggregates plus a bounded // ring of raw request records, all guarded by one mutex. type Stats struct { - mu sync.Mutex - active int64 - byKey map[string]*Stat - byModel map[string]*Stat - bySrc map[string]*Stat - byKeyModel map[string]map[string]*Stat - byKeySrc map[string]map[string]*Stat - byStatus map[int]*Stat // per http status code aggregates (incl. 402/400) - recs []Req - maxRecs int - auditPath string - modelHour map[string]map[int64]int64 // model -> unix-hour bucket -> tokens + mu sync.Mutex + active int64 + byKey map[string]*Stat + byModel map[string]*Stat + bySrc map[string]*Stat + byKeyModel map[string]map[string]*Stat + byKeySrc map[string]map[string]*Stat + byStatus map[int]*Stat // per http status code aggregates (incl. 402/400) + recs []Req + maxRecs int + auditPath string + replayPartial bool // aggregates built from a bounded audit tail + modelHour map[string]map[int64]int64 // model -> unix-hour bucket -> tokens } const hourSec = 3600 @@ -87,14 +91,36 @@ const hourSec = 3600 // file is renamed to ..old and a fresh one is started); pruning // keeps at most auditKeepOld rotated files. Both are vars so tests can shrink // the threshold. +// +// The single-file threshold is deliberately small relative to the total budget +// (16 MB x 16 = 256 MB): records are paged by reading a file backwards, and a +// smaller file keeps the first page from seeking through a huge one. var ( - auditRotateBytes int64 = 64 << 20 - auditKeepOld = 10 + auditRotateBytes int64 = 16 << 20 + auditKeepOld = 16 ) +// auditReplayBytes bounds how much of the audit tail is replayed at startup. +// Replaying the WHOLE file used to cost ~25 MB of resident memory for a 29 MB +// audit log, which dominated the process RSS. Only the recent window belongs in +// memory: older records stay on disk and are paged in on demand by AuditPage / +// StreamAuditRecords, while long-term totals come from the aggregate snapshot +// instead of a full replay. +var auditReplayBytes int64 = 4 << 20 + +// defaultRingSize is how many recent request records stay resident. It has to +// cover two consumers: the status page's 5-minute source windows +// (SourceRecent/SourceAverages) and the first screen of the records table. +// Everything beyond that is paged from the audit file. +const defaultRingSize = 500 + +// firstScreenRecords is how many records a dashboard load ships by default — one +// screen's worth. Scrolling pulls the rest through /api/stats/records. +const firstScreenRecords = 100 + func NewStats(maxRecords int) *Stats { if maxRecords <= 0 { - maxRecords = 3000 + maxRecords = defaultRingSize } return &Stats{ byKey: map[string]*Stat{}, @@ -149,37 +175,72 @@ func incStatus(a *Stat, name string, r Req) { } } +// LoadAudit primes the in-memory state from the audit file. Only the last +// auditReplayBytes are replayed: the ring buffer and the recent-window +// aggregates need the tail, and paging older records is what AuditPage and +// StreamAuditRecords are for. This keeps startup memory proportional to the +// replay window instead of to the (unbounded) audit file. 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 || 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) - } - 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 s.mu.Unlock() + + f, err := os.Open(path) + if err != nil { + return + } + defer f.Close() + fi, err := f.Stat() + if err != nil { + return + } + offset := int64(0) + partial := false + if fi.Size() > auditReplayBytes { + offset = fi.Size() - auditReplayBytes + partial = true // the first line is very likely cut in half + } + if _, err := f.Seek(offset, io.SeekStart); err != nil { + return + } + + s.mu.Lock() + defer s.mu.Unlock() + var recs []Req + sc := bufio.NewScanner(f) + // tolerate long error summaries / oversized junk lines + sc.Buffer(make([]byte, 64*1024), 16*1024*1024) + first := true + for sc.Scan() { + if first { + first = false + if partial { + continue // drop the truncated first record + } + } + var r Req + if json.Unmarshal(sc.Bytes(), &r) != nil || r.Type == "" { + continue // access/event rows (obj) and malformed lines are no requests + } + s.aggregateLocked(r) + recs = append(recs, r) + if len(recs) > s.maxRecs { + // keep the replay bounded in memory as well as on disk: aggregates + // already absorbed the dropped record + recs = recs[len(recs)-s.maxRecs:] + } + } + s.recs = recs + s.replayPartial = partial +} + +// ReplayPartial reports whether the resident aggregates were built from a +// bounded tail of the audit file rather than its full history, so the UI can say +// so instead of implying the numbers are all-time totals. +func (s *Stats) ReplayPartial() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.replayPartial } // Record appends a finished request to the aggregates and ring buffer. @@ -399,9 +460,64 @@ type StatsRow struct { // AuditRecords returns every request row from the audit file plus its rotated // .old files within the [from,to] unix-millisecond window, optionally for one -// masked key id. Unlike Records it is not bounded by the in-memory ring, so it -// yields a complete history for CSV export regardless of restarts or churn. +// masked key id, sorted oldest-first. +// +// It materializes the whole window, so it is only for tests and small windows. +// Production paths must use StreamAuditRecords (O(1) memory, for CSV export) or +// AuditPage (bounded pages, for the records view). func (s *Stats) AuditRecords(from, to int64, key string) []Req { + var out []Req + _ = s.StreamAuditRecords(from, to, key, func(r Req) error { + out = append(out, r) + return nil + }) + sort.Slice(out, func(i, j int) bool { return out[i].Time < out[j].Time }) + return out +} + +// ---- on-demand audit paging ---- +// +// The records view is paged straight off disk instead of being held in memory: +// the newest page is read from the tail of the newest audit file backwards, and +// each response hands back a cursor for the next page. Nothing is cached between +// requests, so "release when the user leaves the page" is guaranteed by never +// retaining anything in the first place. + +// auditCursor points at a byte offset inside one audit file. Paging walks +// backwards, so the cursor means "continue reading BEFORE this offset". +type auditCursor struct { + File string + Offset int64 +} + +// String encodes the cursor for the wire as ":". +func (c auditCursor) String() string { + if c.File == "" { + return "" + } + return c.File + ":" + strconv.FormatInt(c.Offset, 10) +} + +// parseAuditCursor decodes a wire cursor. An empty or malformed value means +// "start at the newest record". +func parseAuditCursor(s string) (auditCursor, bool) { + if s == "" { + return auditCursor{}, false + } + i := strings.LastIndex(s, ":") + if i <= 0 || i == len(s)-1 { + return auditCursor{}, false + } + off, err := strconv.ParseInt(s[i+1:], 10, 64) + if err != nil || off < 0 { + return auditCursor{}, false + } + return auditCursor{File: s[:i], Offset: off}, true +} + +// auditChain lists the audit files newest-first: the live file, then the rotated +// ones in descending timestamp order. +func (s *Stats) auditChain() []string { s.mu.Lock() path := s.auditPath s.mu.Unlock() @@ -410,36 +526,216 @@ func (s *Stats) AuditRecords(from, to int64, key string) []Req { } files := []string{path} if olds, err := filepath.Glob(path + ".*.old"); err == nil { + sort.Sort(sort.Reverse(sort.StringSlice(olds))) files = append(files, olds...) } - var out []Req - for _, p := range files { - f, err := os.Open(p) + return files +} + +// AuditPageResult is one page of records plus the cursor to continue from. +type AuditPageResult struct { + Records []Req `json:"records"` + Next string `json:"next_cursor,omitempty"` + HasMore bool `json:"has_more"` + // Rotated marks that the requested cursor's file no longer exists (the + // audit log rotated under the reader), so the client should restart from + // the newest page instead of silently skipping records. + Rotated bool `json:"rotated,omitempty"` +} + +// revChunk is how much is read per backwards seek. Lines are far shorter than +// this, so a page is typically satisfied by one or two chunks. +const revChunk = 64 * 1024 + +// AuditPage returns up to limit request records ending at cursor, newest first. +// It reads the audit files backwards and stops as soon as the page is full, so +// cost and memory are proportional to limit rather than to the file size. The +// file handle is closed before returning: no state is kept between calls. +func (s *Stats) AuditPage(cursor string, limit int, key string) AuditPageResult { + if limit <= 0 || limit > 1000 { + limit = 100 + } + chain := s.auditChain() + if len(chain) == 0 { + return AuditPageResult{Records: []Req{}} + } + + start := 0 + var offset int64 = -1 // -1 = start at EOF + rotated := false + if c, ok := parseAuditCursor(cursor); ok { + idx := -1 + for i, f := range chain { + if f == c.File { + idx = i + break + } + } + if idx < 0 { + // the cursor's file rotated away: restart from the newest page and + // tell the client so it can reset its view + rotated = true + } else { + start, offset = idx, c.Offset + } + } + + out := make([]Req, 0, limit) + for i := start; i < len(chain); i++ { + next, err := s.pageFile(chain[i], offset, limit, key, &out) + offset = -1 // subsequent files always start at their EOF if err != nil { continue } - 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 || r.Type == "" { - continue + if len(out) >= limit { + res := AuditPageResult{Records: out, Rotated: rotated} + if next > 0 { + res.Next = auditCursor{File: chain[i], Offset: next}.String() + res.HasMore = true + } else if i+1 < len(chain) { + res.Next = auditCursor{File: chain[i+1], Offset: fileSize(chain[i+1])}.String() + res.HasMore = true } - if key != "" && r.Key != key { - continue - } - if from > 0 && r.Time < from { - continue - } - if to > 0 && r.Time > to { - continue - } - out = append(out, r) + return res } - f.Close() } - sort.Slice(out, func(i, j int) bool { return out[i].Time < out[j].Time }) - return out + return AuditPageResult{Records: out, HasMore: false, Rotated: rotated} +} + +func fileSize(path string) int64 { + fi, err := os.Stat(path) + if err != nil { + return 0 + } + return fi.Size() +} + +// pageFile reads one audit file backwards from end (or EOF when end < 0), +// appending matching records to out (newest first) until it holds limit entries. +// It returns the offset to continue from within this file, or 0 when the file is +// exhausted. +func (s *Stats) pageFile(path string, end int64, limit int, key string, out *[]Req) (int64, error) { + f, err := os.Open(path) + if err != nil { + return 0, err + } + defer f.Close() // released before the response is written: nothing is retained + fi, err := f.Stat() + if err != nil { + return 0, err + } + if end < 0 || end > fi.Size() { + end = fi.Size() + } + + buf := make([]byte, 0, revChunk) + var tail []byte // bytes of a line whose start was not in this chunk + pos := end + for pos > 0 && len(*out) < limit { + size := int64(revChunk) + if size > pos { + size = pos + } + pos -= size + buf = buf[:size] + if _, err := f.ReadAt(buf, pos); err != nil && err != io.EOF { + return 0, err + } + chunk := buf + if len(tail) > 0 { + chunk = append(append(make([]byte, 0, len(buf)+len(tail)), buf...), tail...) + tail = tail[:0] + } + // walk the chunk's complete lines from the end towards the start + lineEnd := len(chunk) + for i := len(chunk) - 1; i >= 0; i-- { + if chunk[i] != '\n' { + continue + } + if rec, ok := decodeAuditLine(chunk[i+1:lineEnd], key); ok { + *out = append(*out, rec) + if len(*out) >= limit { + // resume at the newline we just consumed + return pos + int64(i) + 1, nil + } + } + lineEnd = i + } + // chunk[:lineEnd] is an incomplete line: carry it into the next chunk + if lineEnd > 0 { + tail = append(tail[:0], chunk[:lineEnd]...) + } + if pos == 0 { + // start of file: whatever is carried is a complete first line + if rec, ok := decodeAuditLine(tail, key); ok { + *out = append(*out, rec) + } + } + } + if pos <= 0 { + return 0, nil // file exhausted + } + return pos, nil +} + +// decodeAuditLine parses one audit line as a request record, filtering by key. +// Access/event rows and malformed lines are rejected. +func decodeAuditLine(line []byte, key string) (Req, bool) { + line = bytes.TrimSpace(line) + if len(line) == 0 { + return Req{}, false + } + var r Req + if json.Unmarshal(line, &r) != nil || r.Type == "" { + return Req{}, false + } + if key != "" && r.Key != key { + return Req{}, false + } + return r, true +} + +// StreamAuditRecords walks every request record in [from, to] (unix millis, +// 0 = unbounded) newest file first and hands each one to fn. Nothing is +// accumulated: an export of an arbitrarily long period costs O(1) memory, and +// the file handles are closed as the walk proceeds. fn returning an error stops +// the walk (used to abort on a broken client connection). +func (s *Stats) StreamAuditRecords(from, to int64, key string, fn func(Req) error) error { + for _, path := range s.auditChain() { + if err := streamAuditFile(path, from, to, key, fn); err != nil { + return err + } + } + return nil +} + +func streamAuditFile(path string, from, to int64, key string, fn func(Req) error) error { + f, err := os.Open(path) + if err != nil { + return nil // a rotated-away file is not an export failure + } + defer f.Close() + 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 || r.Type == "" { + continue + } + if key != "" && r.Key != key { + continue + } + if from > 0 && r.Time < from { + continue + } + if to > 0 && r.Time > to { + continue + } + if err := fn(r); err != nil { + return err + } + } + return nil } // SourceRecent counts real gateway requests per source within the last @@ -530,10 +826,18 @@ func (s *Stats) SourceAverages(windowSec int64) map[string]SourceAvg { // Snapshot returns the whole dashboard payload; when key != "" the records // and aggregate views are restricted to that gateway key. +// Snapshot returns the aggregate rows plus the FIRST SCREEN of request records +// (newest last, matching the ring order). limit bounds the record slice only; +// older records are not included here at all — the dashboard pages them from +// /api/stats/records as the user scrolls, so a dashboard load never serializes +// the whole ring. limit <= 0 means the default first screen. func (s *Stats) Snapshot(limit int, key string) map[string]interface{} { s.mu.Lock() defer s.mu.Unlock() if limit <= 0 { + limit = firstScreenRecords + } + if limit > s.maxRecs { limit = s.maxRecs } start := 0 @@ -581,12 +885,13 @@ func (s *Stats) Snapshot(limit int, key string) map[string]interface{} { return ci < cj }) return map[string]interface{}{ - "active": s.active, - "total": total, - "by_key": rows(byKey), - "by_model": rows(byModel), - "by_source": rows(bySrc), - "by_status": bs, - "records": append([]Req(nil), recs...), + "active": s.active, + "total": total, + "by_key": rows(byKey), + "by_model": rows(byModel), + "by_source": rows(bySrc), + "by_status": bs, + "records": append([]Req(nil), recs...), + "replay_partial": s.replayPartial, } } diff --git a/internal/gateway/stats_test.go b/internal/gateway/stats_test.go index 4460af2..4f23c1d 100644 --- a/internal/gateway/stats_test.go +++ b/internal/gateway/stats_test.go @@ -156,4 +156,347 @@ func TestLoadAuditFullReplay(t *testing.T) { 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 +} + +// ---- on-demand log loading (plan 阶段 3) ---- + +// writeAuditFile writes n synthetic request rows (plus some non-request event +// rows, which must be skipped) and returns the path. +func writeAuditFile(t *testing.T, path string, n int, key string, startMs int64) { + t.Helper() + f, err := os.Create(path) + if err != nil { + t.Fatalf("create audit: %v", err) + } + defer f.Close() + for i := 0; i < n; i++ { + fmt.Fprintf(f, `{"time":%d,"key":%q,"type":"chat","model":"m","source":"s","prompt_tokens":1,"completion_tokens":1,"latency_ms":5,"ok":true,"status":200}`+"\n", + startMs+int64(i)*1000, key) + if i%10 == 0 { + // an access-log/event row: must never appear as a record + fmt.Fprintf(f, `{"obj":"access","time":%d,"path":"/v1/models"}`+"\n", startMs+int64(i)*1000) + } + } +} + +// TestLoadAuditBoundedTail is the startup-memory fix: only the tail of a large +// audit file is replayed, and the ring stays capped. +func TestLoadAuditBoundedTail(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "audit.jsonl") + writeAuditFile(t, path, 20000, "k1", 1_700_000_000_000) + + fi, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + orig := auditReplayBytes + auditReplayBytes = 64 << 10 // 64 KB tail out of a much larger file + defer func() { auditReplayBytes = orig }() + if fi.Size() <= auditReplayBytes { + t.Fatalf("fixture too small (%d bytes) to exercise bounded replay", fi.Size()) + } + + s := NewStats(0) + s.LoadAudit(path) + if !s.ReplayPartial() { + t.Fatal("a truncated replay must be reported as partial") + } + snap := s.Snapshot(0, "") + recs, _ := snap["records"].([]Req) + if len(recs) == 0 { + t.Fatal("bounded replay must still load the recent window") + } + if len(recs) > firstScreenRecords { + t.Fatalf("snapshot shipped %d records, must be capped at the first screen (%d)", + len(recs), firstScreenRecords) + } + // the aggregates must come from the tail only, not the whole file + total, _ := snap["total"].(Stat) + if total.Reqs == 0 { + t.Fatal("tail replay must feed the aggregates") + } + if total.Reqs >= 20000 { + t.Fatalf("aggregates replayed the whole file (%d reqs); replay must be bounded", total.Reqs) + } +} + +// TestLoadAuditFullFileWhenSmall: a file inside the replay window is loaded +// completely and is not flagged partial. +func TestLoadAuditFullFileWhenSmall(t *testing.T) { + path := filepath.Join(t.TempDir(), "audit.jsonl") + writeAuditFile(t, path, 50, "k1", 1_700_000_000_000) + s := NewStats(0) + s.LoadAudit(path) + if s.ReplayPartial() { + t.Fatal("a small file must be replayed in full, not flagged partial") + } + snap := s.Snapshot(0, "") + total, _ := snap["total"].(Stat) + if total.Reqs != 50 { + t.Fatalf("total reqs = %d, want 50", total.Reqs) + } +} + +// TestQuotaWindowsSurviveBoundedReplay is the regression guard the plan calls +// out: shrinking the ring must not break token-quota accounting, which reads the +// hour buckets rather than the ring. +func TestQuotaWindowsSurviveBoundedReplay(t *testing.T) { + s := NewStats(10) // deliberately tiny ring + now := time.Now().UnixMilli() + for i := 0; i < 100; i++ { + s.Record(Req{Time: now, Key: "k", Model: "m", Source: "s", Prompt: 2, Compl: 3, OK: true, Status: 200}) + } + if got := s.WindowTokens("m", "s", 24*hourSec); got != 500 { + t.Fatalf("window tokens = %d, want 500 (quota accounting must not depend on the ring)", got) + } + if got := s.ModelTokens("k")["m"]; got != 500 { + t.Fatalf("model tokens = %d, want 500", got) + } + if got := s.KeyTokens("k"); got != 500 { + t.Fatalf("key tokens = %d, want 500", got) + } +} + +// TestAuditPagePagesBackwards walks the whole history one page at a time and +// asserts strict newest-first order with no gaps or repeats. +func TestAuditPagePagesBackwards(t *testing.T) { + path := filepath.Join(t.TempDir(), "audit.jsonl") + const n = 250 + writeAuditFile(t, path, n, "k1", 1_700_000_000_000) + s := NewStats(0) + s.LoadAudit(path) + + var seen []Req + cursor := "" + for i := 0; i < 20; i++ { // bounded loop guards against a cursor that never advances + page := s.AuditPage(cursor, 40, "") + if len(page.Records) == 0 { + break + } + seen = append(seen, page.Records...) + if !page.HasMore { + break + } + if page.Next == cursor { + t.Fatal("cursor did not advance") + } + cursor = page.Next + } + if len(seen) != n { + t.Fatalf("paged %d records, want %d", len(seen), n) + } + for i := 1; i < len(seen); i++ { + if seen[i-1].Time < seen[i].Time { + t.Fatalf("records not newest-first at %d: %d then %d", i, seen[i-1].Time, seen[i].Time) + } + } + // newest record first, oldest last + if seen[0].Time != 1_700_000_000_000+int64(n-1)*1000 { + t.Fatalf("first record time = %d, want the newest", seen[0].Time) + } + if seen[len(seen)-1].Time != 1_700_000_000_000 { + t.Fatalf("last record time = %d, want the oldest", seen[len(seen)-1].Time) + } +} + +// TestAuditPageFirstScreenIsSmall: the first page must be exactly one screen, +// independent of how large the audit file is. +func TestAuditPageFirstScreenIsSmall(t *testing.T) { + path := filepath.Join(t.TempDir(), "audit.jsonl") + writeAuditFile(t, path, 5000, "k1", 1_700_000_000_000) + s := NewStats(0) + s.LoadAudit(path) + page := s.AuditPage("", 100, "") + if len(page.Records) != 100 { + t.Fatalf("first page = %d records, want 100", len(page.Records)) + } + if !page.HasMore || page.Next == "" { + t.Fatal("a large file must report more pages plus a cursor") + } +} + +// TestAuditPageFiltersByKey: a user key may only page its own records. +func TestAuditPageFiltersByKey(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "audit.jsonl") + writeAuditFile(t, path, 30, "mine", 1_700_000_000_000) + f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0644) + if err != nil { + t.Fatal(err) + } + for i := 0; i < 30; i++ { + fmt.Fprintf(f, `{"time":%d,"key":"other","type":"chat","model":"m","source":"s","ok":true,"status":200}`+"\n", + 1_700_000_100_000+int64(i)*1000) + } + f.Close() + + s := NewStats(0) + s.LoadAudit(path) + page := s.AuditPage("", 1000, "mine") + if len(page.Records) != 30 { + t.Fatalf("filtered page = %d records, want 30", len(page.Records)) + } + for _, r := range page.Records { + if r.Key != "mine" { + t.Fatalf("key filter leaked record for %q", r.Key) + } + } +} + +// TestAuditPageSpansRotatedFiles: paging must continue into .old files instead +// of stopping at the live file's start. +func TestAuditPageSpansRotatedFiles(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "audit.jsonl") + writeAuditFile(t, path, 40, "k1", 1_700_000_500_000) // newest + writeAuditFile(t, path+".100.old", 40, "k1", 1_700_000_300_000) // older + writeAuditFile(t, path+".050.old", 40, "k1", 1_700_000_100_000) // oldest + + s := NewStats(0) + s.LoadAudit(path) + var seen []Req + cursor := "" + for i := 0; i < 30; i++ { + page := s.AuditPage(cursor, 25, "") + seen = append(seen, page.Records...) + if !page.HasMore { + break + } + cursor = page.Next + } + if len(seen) != 120 { + t.Fatalf("paged %d records across rotated files, want 120", len(seen)) + } + for i := 1; i < len(seen); i++ { + if seen[i-1].Time < seen[i].Time { + t.Fatalf("rotation boundary broke the ordering at %d", i) + } + } +} + +// TestAuditPageRotatedCursor: a cursor whose file rotated away must restart from +// the newest page and say so, rather than silently returning nothing. +func TestAuditPageRotatedCursor(t *testing.T) { + path := filepath.Join(t.TempDir(), "audit.jsonl") + writeAuditFile(t, path, 20, "k1", 1_700_000_000_000) + s := NewStats(0) + s.LoadAudit(path) + page := s.AuditPage("/gone/audit.jsonl:12345", 10, "") + if !page.Rotated { + t.Fatal("a stale cursor must be reported as rotated") + } + if len(page.Records) == 0 { + t.Fatal("a stale cursor must fall back to the newest page") + } +} + +// TestAuditPageEmptyAndNoFile cover the degenerate inputs. +func TestAuditPageEmptyAndNoFile(t *testing.T) { + s := NewStats(0) + if page := s.AuditPage("", 10, ""); len(page.Records) != 0 || page.HasMore { + t.Fatalf("no audit path must yield an empty page, got %+v", page) + } + path := filepath.Join(t.TempDir(), "audit.jsonl") + if err := os.WriteFile(path, nil, 0644); err != nil { + t.Fatal(err) + } + s2 := NewStats(0) + s2.LoadAudit(path) + if page := s2.AuditPage("", 10, ""); len(page.Records) != 0 || page.HasMore { + t.Fatalf("empty audit file must yield an empty page, got %+v", page) + } +} + +// TestAuditPageHandlesLongLines: a record larger than the reverse-read chunk +// must still be decoded (the reader carries partial lines across chunks). +func TestAuditPageHandlesLongLines(t *testing.T) { + path := filepath.Join(t.TempDir(), "audit.jsonl") + f, err := os.Create(path) + if err != nil { + t.Fatal(err) + } + big := strings.Repeat("e", revChunk+1024) // one record spanning >1 chunk + fmt.Fprintf(f, `{"time":1,"key":"k","type":"chat","model":"m","source":"s","ok":false,"status":500,"error":%q}`+"\n", big) + fmt.Fprintf(f, `{"time":2,"key":"k","type":"chat","model":"m","source":"s","ok":true,"status":200}`+"\n") + f.Close() + + s := NewStats(0) + s.LoadAudit(path) + page := s.AuditPage("", 10, "") + if len(page.Records) != 2 { + t.Fatalf("got %d records, want 2 (a >chunk record must survive reverse reads)", len(page.Records)) + } + if page.Records[0].Time != 2 || page.Records[1].Time != 1 { + t.Fatalf("ordering broken: %d then %d", page.Records[0].Time, page.Records[1].Time) + } + if len(page.Records[1].Err) != len(big) { + t.Fatalf("oversized field truncated: %d bytes, want %d", len(page.Records[1].Err), len(big)) + } +} + +// TestStreamAuditRecordsIsWindowed: export streaming must honour the time window +// and the key filter, and must hand records over one at a time. +func TestStreamAuditRecordsIsWindowed(t *testing.T) { + path := filepath.Join(t.TempDir(), "audit.jsonl") + writeAuditFile(t, path, 100, "k1", 1_700_000_000_000) + s := NewStats(0) + s.LoadAudit(path) + + from := int64(1_700_000_010_000) + to := int64(1_700_000_019_000) + var got []Req + if err := s.StreamAuditRecords(from, to, "", func(r Req) error { + got = append(got, r) + return nil + }); err != nil { + t.Fatalf("stream: %v", err) + } + if len(got) != 10 { + t.Fatalf("streamed %d records, want 10 in [%d,%d]", len(got), from, to) + } + for _, r := range got { + if r.Time < from || r.Time > to { + t.Fatalf("record outside the window: %d", r.Time) + } + } +} + +// TestStreamAuditRecordsAbortStops: a callback error (client disconnect) must +// stop the walk immediately instead of reading the rest of the history. +func TestStreamAuditRecordsAbortStops(t *testing.T) { + path := filepath.Join(t.TempDir(), "audit.jsonl") + writeAuditFile(t, path, 500, "k1", 1_700_000_000_000) + s := NewStats(0) + s.LoadAudit(path) + + seen := 0 + stop := fmt.Errorf("client gone") + err := s.StreamAuditRecords(0, 0, "", func(Req) error { + seen++ + if seen == 5 { + return stop + } + return nil + }) + if err != stop { + t.Fatalf("abort must surface the callback error, got %v", err) + } + if seen != 5 { + t.Fatalf("walk continued after abort: %d records", seen) + } +} + +// TestAuditRotationThresholds pins the smaller-file/more-files trade-off that +// makes reverse paging cheap while keeping the same total budget. +func TestAuditRotationThresholds(t *testing.T) { + if auditRotateBytes != 16<<20 { + t.Fatalf("auditRotateBytes = %d, want 16 MiB", auditRotateBytes) + } + if auditKeepOld != 16 { + t.Fatalf("auditKeepOld = %d, want 16", auditKeepOld) + } + if total := auditRotateBytes * int64(auditKeepOld); total != 256<<20 { + t.Fatalf("total audit budget = %d, want 256 MiB", total) + } +}