mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-09-20 17:07:59 +00:00
perf(gateway): load request logs on demand instead of holding them in memory
Startup RSS on this deployment was 56 MB with a 29 MB audit log and ~10 MB
without one: LoadAudit() json-unmarshalled the ENTIRE file into the aggregates
and kept a 10000-entry ring of raw records. Two more paths had the same shape —
AuditRecords() materialized a whole export window into a []Req before sorting
it, and a dashboard poll serialized the full ring so the browser could render
300 rows of it.
The audit file is now the source of truth and memory only holds the live
window:
* LoadAudit replays only the last auditReplayBytes (4 MB) and drops the
truncated first line; the ring default drops 10000 -> 500, which still
covers both of its consumers (the status page's 5-minute SourceRecent /
SourceAverages windows and the first screen of the records table).
replayPartial is exported so the UI can say the totals cover a window
rather than all time. Token-quota accounting is unaffected: it reads the
modelHour buckets, not the ring (pinned by a test).
* AuditPage(cursor, limit, key) pages records straight off disk, reading the
newest file backwards in 64 KB chunks and returning as soon as the page is
full. The cursor is "<file>:<offset>" and walks into rotated .old files;
a cursor whose file rotated away reports rotated=true so the client can
reset instead of silently skipping records. No state is cached between
requests and the file handle is closed before responding, so "release when
the user leaves the page" is guaranteed by never retaining anything.
* StreamAuditRecords(from,to,key,fn) replaces the accumulate-then-sort export
path; the CSV handler writes rows as they are read and flushes every 1000,
and a write error (client gone) aborts the walk. Export memory is O(1)
regardless of the window. AuditRecords is kept as a test-only wrapper.
* Snapshot ships one screen (firstScreenRecords=100) by default; aggregates
are untouched.
* Audit rotation 64 MB x 10 -> 16 MB x 16: same 256 MB total budget, but a
smaller newest file keeps the first reverse page cheap.
New route: GET /api/stats/records?before=&limit=&key= (non-admins are pinned to
their own key by exportKey). /api/status additionally reports adapter_pools for
admins.
Measured with production's 29 MB audit copied to the test instance: startup RSS
19.0 MB (was 56 MB); scrolling 10 pages (1000 records) +0.7 MB; exporting the
full history (36441 rows / 4.4 MB CSV) +0.1 MB with no residual growth.
This commit is contained in:
JianFeeeee
@ -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=<cursor>&limit=100[&key=<masked id>]
|
||||
// -> { 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,
|
||||
})
|
||||
}
|
||||
|
||||
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
@ -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 <path>.<unix>.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 "<file>:<offset>".
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- 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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user