diff --git a/cmd/server/config.go b/cmd/server/config.go index d784ddf5..54c7c0a8 100644 --- a/cmd/server/config.go +++ b/cmd/server/config.go @@ -26,6 +26,12 @@ type Config struct { // （issue #41：截断的 JSON 让上游 unmarshal 报 unexpected EOF，网关却罚号）。 // 0/负数视为非法 → normalize 回落默认并记录。 MaxBodyMB int `json:"max_body_mb"` + + // MetricsEnabled 是否采集按模型的请求统计（默认 true）。 + MetricsEnabled bool `json:"metrics_enabled"` + // MetricsFile 统计持久化文件；空 = 纯内存（重启清零）。 + // 默认 ./data/metrics.json，重启后累计值不丢。 + MetricsFile string `json:"metrics_file"` } `json:"server"` Cooldown struct { @@ -152,6 +158,8 @@ func Default() *Config { c.Cooldown.SoftRate = "600s" c.Cooldown.SoftRateMax = "2h" c.Server.MaxBodyMB = 8 // 请求体上限默认 8MB + c.Server.MetricsEnabled = true + c.Server.MetricsFile = "./data/metrics.json" // 排程段默认值由 internal/config 集中维护（cmd/server 与 cmd/activity 共用， // 消除 issue #49 的默认值漂移）。 c.Schedule = config.DefaultSchedule() diff --git a/cmd/server/main.go b/cmd/server/main.go index be2e9162..f95d4e88 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -12,6 +12,7 @@ import ( "time" "workbuddy2api/internal/auth" + "workbuddy2api/internal/metrics" "workbuddy2api/internal/pool" "workbuddy2api/internal/redisstore" "workbuddy2api/internal/scheduler" @@ -171,7 +172,14 @@ func main() { log.Printf("夜猫子任务已启用：%v 点（task_runner.py ALL --yes --only black_cat）", cfg.Schedule.CatHours) } + var metricsCollector *metrics.Collector + if cfg.Server.MetricsEnabled { + metricsCollector = metrics.New(cfg.Server.MetricsFile) + defer metricsCollector.Flush() // 退出前落盘 + } + h := server.NewHandler(server.Config{ + Metrics: metricsCollector, Pool: p, Upstream: up, APIKey: cfg.APIKey, diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go new file mode 100644 index 00000000..b0fd2df7 --- /dev/null +++ b/internal/metrics/metrics.go @@ -0,0 +1,432 @@ +// Package metrics 按模型维度的请求量/token/缓存命中/扣费统计。 +// +// 设计取态：网关是唯一能看到**所有**请求（含绕过面板的其他客户端）的位置， +// 因此统计在这里采集，通过 /v1/stats 暴露给面板。 +// +// 为什么按模型分组：不同模型的定价、上下文长度、缓存行为差异极大 +// （cache 命中率直接影响实际扣费），混在一起看没有决策价值。 +package metrics + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "sync" + "time" +) + +// ModelStats 单个模型的累计统计。 +// +// 所有计数都是**累计值**（进程启动至今），面板取两次快照的差值即可算速率。 +type ModelStats struct { + Model string `json:"model"` + + // ── 请求量 ────────────────────────────────────────── + Requests int64 `json:"requests"` // 总请求数 + Success int64 `json:"success"` // 2xx + Failed int64 `json:"failed"` // 非 2xx / 传输失败 + Streaming int64 `json:"streaming"` // 其中流式请求数 + + // ── 延迟（毫秒，累计和，由面板算平均）────────────── + // 用累计和而非滑动窗口：无需后台 goroutine，重启后仍能从持久化恢复。 + TTFBSumMS int64 `json:"ttfb_sum_ms"` // 首字延迟累加（仅流式有值） + TTFBCount int64 `json:"ttfb_count"` // 有 TTFB 采样的请求数 + LatencySumMS int64 `json:"latency_sum_ms"` // 端到端耗时累加 + LatencyCount int64 `json:"latency_count"` + + // ── Token ────────────────────────────────────────── + PromptTokens int64 `json:"prompt_tokens"` + CompletionTokens int64 `json:"completion_tokens"` + TotalTokens int64 `json:"total_tokens"` + UsageReported int64 `json:"usage_reported"` // 带 usage 的请求数（算平均值用） + + // ── 缓存（prompt cache）──────────────────────────── + // 上游按 prompt_cache_hit/miss 区分计费，命中部分通常便宜得多。 + CacheHitTokens int64 `json:"cache_hit_tokens"` + CacheMissTokens int64 `json:"cache_miss_tokens"` + CacheWriteTokens int64 `json:"cache_write_tokens"` + // CacheReadTokens/CacheCreationTokens 是另一套命名（部分模型用），一并记录。 + CacheReadTokens int64 `json:"cache_read_tokens"` + CacheCreationTokens int64 `json:"cache_creation_tokens"` + + // ── 扣费 ─────────────────────────────────────────── + // CreditMilli 用「毫」为单位累计（上游 credit 是两位小数），避免浮点误差。 + CreditMilli int64 `json:"credit_milli"` + + // ── 吞吐（由面板按 token/耗时算）─────────────────── + // 这里额外记录"有首字到结束"的时长，用于算纯生成速率（剔除排队等待）。 + GenSumMS int64 `json:"gen_sum_ms"` // 首字→结束的毫秒累加 + GenCount int64 `json:"gen_count"` + + FirstSeen time.Time `json:"first_seen"` + LastSeen time.Time `json:"last_seen"` +} + +// Snapshot 一次采样的完整视图。 +type Snapshot struct { + // Models 按模型名索引的累计统计。 + Models map[string]*ModelStats `json:"models"` + // Total 全模型汇总（便于面板直接展示总量）。 + Total ModelStats `json:"total"` + // Since 统计起点（进程启动或上次重置）。 + Since time.Time `json:"since"` + // Now 快照时刻（面板用它和 Since 算运行时长）。 + Now time.Time `json:"now"` +} + +// delta 单个请求的观测值，由 handler 填充。 +type Delta struct { + Model string + Stream bool + OK bool + TTFB time.Duration // 流式首字延迟；同步请求为 0 + Latency time.Duration // 端到端 + HasUsage bool + + PromptTokens int64 + CompletionTokens int64 + TotalTokens int64 + + CacheHitTokens int64 + CacheMissTokens int64 + CacheWriteTokens int64 + CacheReadTokens int64 + CacheCreationTokens int64 + + Credit float64 +} + +// Collector 线程安全的统计收集器。 +type Collector struct { + mu sync.Mutex + models map[string]*ModelStats + since time.Time + + // stateFile 非空时落盘（重启后累计值不丢）。 + stateFile string + // dirty 有未落盘变更。 + dirty bool + // flushEvery 每 N 次记录落盘一次（避免每个请求都写盘）。 + flushEvery int + sinceFlush int +} + +// New 构建收集器；stateFile 为空表示纯内存（不持久化）。 +func New(stateFile string) *Collector { + c := &Collector{ + models: map[string]*ModelStats{}, + since: time.Now(), + stateFile: stateFile, + flushEvery: 20, + } + if stateFile != "" { + c.load() + } + return c +} + +// Record 记录一次请求。 +func (c *Collector) Record(d Delta) { + model := d.Model + if model == "" { + model = "(unknown)" + } + now := time.Now() + + c.mu.Lock() + defer c.mu.Unlock() + + m, ok := c.models[model] + if !ok { + m = &ModelStats{Model: model, FirstSeen: now} + c.models[model] = m + } + m.LastSeen = now + + m.Requests++ + if d.OK { + m.Success++ + } else { + m.Failed++ + } + if d.Stream { + m.Streaming++ + } + + if d.TTFB > 0 { + m.TTFBSumMS += d.TTFB.Milliseconds() + m.TTFBCount++ + } + if d.Latency > 0 { + ms := d.Latency.Milliseconds() + m.LatencySumMS += ms + m.LatencyCount++ + // 生成时长 = 端到端 - 首字等待（流式才有意义）。 + if d.TTFB > 0 && d.Latency > d.TTFB { + m.GenSumMS += (d.Latency - d.TTFB).Milliseconds() + m.GenCount++ + } + } + + if d.HasUsage { + m.UsageReported++ + m.PromptTokens += d.PromptTokens + m.CompletionTokens += d.CompletionTokens + m.TotalTokens += d.TotalTokens + m.CacheHitTokens += d.CacheHitTokens + m.CacheMissTokens += d.CacheMissTokens + m.CacheWriteTokens += d.CacheWriteTokens + m.CacheReadTokens += d.CacheReadTokens + m.CacheCreationTokens += d.CacheCreationTokens + } + // credit 以「毫」累计：上游给两位小数（0.02），×1000 后是整数。 + if d.Credit != 0 { + m.CreditMilli += int64(d.Credit*1000 + 0.5) + } + + c.dirty = true + c.sinceFlush++ + if c.stateFile != "" && c.sinceFlush >= c.flushEvery { + c.saveLocked() + } +} + +// Snapshot 返回当前累计统计的深拷贝（含全模型汇总）。 +func (c *Collector) Snapshot() Snapshot { + c.mu.Lock() + defer c.mu.Unlock() + + out := Snapshot{ + Models: make(map[string]*ModelStats, len(c.models)), + Since: c.since, + Now: time.Now(), + } + total := &ModelStats{Model: "(all)"} + for name, m := range c.models { + cp := *m + out.Models[name] = &cp + addInto(total, &cp) + } + out.Total = *total + return out +} + +// Reset 清空统计（面板"重置统计"用）。 +func (c *Collector) Reset() { + c.mu.Lock() + defer c.mu.Unlock() + c.models = map[string]*ModelStats{} + c.since = time.Now() + c.dirty = true + c.saveLocked() +} + +// Flush 强制落盘（进程退出前调用）。 +func (c *Collector) Flush() { + c.mu.Lock() + defer c.mu.Unlock() + if c.dirty { + c.saveLocked() + } +} + +// addInto 把 src 的计数累加进 dst（汇总用）。 +func addInto(dst, src *ModelStats) { + dst.Requests += src.Requests + dst.Success += src.Success + dst.Failed += src.Failed + dst.Streaming += src.Streaming + dst.TTFBSumMS += src.TTFBSumMS + dst.TTFBCount += src.TTFBCount + dst.LatencySumMS += src.LatencySumMS + dst.LatencyCount += src.LatencyCount + dst.PromptTokens += src.PromptTokens + dst.CompletionTokens += src.CompletionTokens + dst.TotalTokens += src.TotalTokens + dst.UsageReported += src.UsageReported + dst.CacheHitTokens += src.CacheHitTokens + dst.CacheMissTokens += src.CacheMissTokens + dst.CacheWriteTokens += src.CacheWriteTokens + dst.CacheReadTokens += src.CacheReadTokens + dst.CacheCreationTokens += src.CacheCreationTokens + dst.CreditMilli += src.CreditMilli + dst.GenSumMS += src.GenSumMS + dst.GenCount += src.GenCount + if dst.FirstSeen.IsZero() || (!src.FirstSeen.IsZero() && src.FirstSeen.Before(dst.FirstSeen)) { + dst.FirstSeen = src.FirstSeen + } + if src.LastSeen.After(dst.LastSeen) { + dst.LastSeen = src.LastSeen + } +} + +// --------------------------------------------------------------------------- +// 持久化 +// --------------------------------------------------------------------------- + +type stateFile struct { + Since time.Time `json:"since"` + Models map[string]*ModelStats `json:"models"` +} + +// saveLocked 原子落盘。调用方必须已持锁。 +// 落盘失败只打日志不阻断（统计是观测功能，不应影响转发）。 +func (c *Collector) saveLocked() { + c.dirty = false + c.sinceFlush = 0 + if c.stateFile == "" { + return + } + raw, err := json.MarshalIndent(stateFile{Since: c.since, Models: c.models}, "", " ") + if err != nil { + return + } + if dir := filepath.Dir(c.stateFile); dir != "" { + _ = os.MkdirAll(dir, 0o755) + } + tmp := c.stateFile + ".tmp" + if err := os.WriteFile(tmp, raw, 0o600); err != nil { + return + } + _ = os.Rename(tmp, c.stateFile) +} + +// load 启动时恢复累计统计（文件缺失/损坏时静默从零开始）。 +func (c *Collector) load() { + raw, err := os.ReadFile(c.stateFile) + if err != nil { + return + } + var sf stateFile + if json.Unmarshal(raw, &sf) != nil { + return + } + if sf.Models != nil { + c.models = sf.Models + } + if !sf.Since.IsZero() { + c.since = sf.Since + } +} + +// --------------------------------------------------------------------------- +// 展示辅助（面板可直接用的派生指标） +// --------------------------------------------------------------------------- + +// Derived 单模型的派生指标（平均/速率），面板表格直接用。 +type Derived struct { + Model string `json:"model"` + + Requests int64 `json:"requests"` + Success int64 `json:"success"` + Failed int64 `json:"failed"` + Streaming int64 `json:"streaming"` + + // AvgTTFBMS 平均首字延迟（毫秒）；无采样为 0。 + AvgTTFBMS float64 `json:"avg_ttfb_ms"` + // AvgLatencyMS 平均端到端耗时（毫秒）。 + AvgLatencyMS float64 `json:"avg_latency_ms"` + // TokensPerSec 生成速率 = 输出 token / 生成秒数（剔除首字等待）。 + TokensPerSec float64 `json:"tokens_per_sec"` + + PromptTokens int64 `json:"prompt_tokens"` + CompletionTokens int64 `json:"completion_tokens"` + TotalTokens int64 `json:"total_tokens"` + + CacheHitTokens int64 `json:"cache_hit_tokens"` + CacheMissTokens int64 `json:"cache_miss_tokens"` + CacheWriteTokens int64 `json:"cache_write_tokens"` + // CacheHitRate 缓存命中率 = hit / (hit + miss)，无数据为 0。 + CacheHitRate float64 `json:"cache_hit_rate"` + + // Credit 累计扣费（元/积分，两位小数）。 + Credit float64 `json:"credit"` + // CreditPerReq 平均每请求扣费。 + CreditPerReq float64 `json:"credit_per_req"` + + LastSeen *time.Time `json:"last_seen,omitempty"` +} + +// Derive 把累计统计换算成派生指标。 +func Derive(m *ModelStats) Derived { + d := Derived{ + Model: m.Model, + Requests: m.Requests, + Success: m.Success, + Failed: m.Failed, + Streaming: m.Streaming, + PromptTokens: m.PromptTokens, + CompletionTokens: m.CompletionTokens, + TotalTokens: m.TotalTokens, + CacheHitTokens: m.CacheHitTokens + m.CacheReadTokens, + CacheMissTokens: m.CacheMissTokens, + CacheWriteTokens: m.CacheWriteTokens + m.CacheCreationTokens, + Credit: float64(m.CreditMilli) / 1000, + } + if m.TTFBCount > 0 { + d.AvgTTFBMS = float64(m.TTFBSumMS) / float64(m.TTFBCount) + } + if m.LatencyCount > 0 { + d.AvgLatencyMS = float64(m.LatencySumMS) / float64(m.LatencyCount) + } + // 生成速率：优先用"首字→结束"的纯生成时长；无 TTFB 采样时退回端到端耗时。 + if m.GenCount > 0 && m.GenSumMS > 0 { + d.TokensPerSec = float64(m.CompletionTokens) / (float64(m.GenSumMS) / 1000) + } else if m.LatencySumMS > 0 { + d.TokensPerSec = float64(m.CompletionTokens) / (float64(m.LatencySumMS) / 1000) + } + if total := d.CacheHitTokens + d.CacheMissTokens; total > 0 { + d.CacheHitRate = float64(d.CacheHitTokens) / float64(total) + } + if m.Requests > 0 { + d.CreditPerReq = d.Credit / float64(m.Requests) + } + if !m.LastSeen.IsZero() { + t := m.LastSeen + d.LastSeen = &t + } + return d +} + +// DerivedSnapshot 面板用的完整派生视图。 +type DerivedSnapshot struct { + Models []Derived `json:"models"` + Total Derived `json:"total"` + Since time.Time `json:"since"` + Now time.Time `json:"now"` + // UptimeSec 统计持续时间（秒），面板算速率用。 + UptimeSec int64 `json:"uptime_sec"` +} + +// Derived 生成面板视图（模型按请求数降序）。 +func (c *Collector) Derived() DerivedSnapshot { + snap := c.Snapshot() + out := DerivedSnapshot{ + Models: make([]Derived, 0, len(snap.Models)), + Since: snap.Since, + Now: snap.Now, + } + for _, m := range snap.Models { + out.Models = append(out.Models, Derive(m)) + } + sort.Slice(out.Models, func(i, j int) bool { + if out.Models[i].Requests != out.Models[j].Requests { + return out.Models[i].Requests > out.Models[j].Requests + } + return out.Models[i].Model < out.Models[j].Model + }) + out.Total = Derive(&snap.Total) + out.Total.Model = "(all)" + if !snap.Since.IsZero() { + out.UptimeSec = int64(snap.Now.Sub(snap.Since).Seconds()) + } + return out +} + +// String 便于日志调试。 +func (d Derived) String() string { + return fmt.Sprintf("%s req=%d ttfb=%.0fms tok/s=%.1f cache=%.0f%% credit=%.2f", + d.Model, d.Requests, d.AvgTTFBMS, d.TokensPerSec, d.CacheHitRate*100, d.Credit) +} diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go new file mode 100644 index 00000000..4bbb5ec4 --- /dev/null +++ b/internal/metrics/metrics_test.go @@ -0,0 +1,267 @@ +package metrics + +import ( + "os" + "path/filepath" + "sync" + "testing" + "time" +) + +// TestRecordAggregatesByModel 统计必须按模型分开累计，且各计数正确。 +func TestRecordAggregatesByModel(t *testing.T) { + c := New("") + + c.Record(Delta{ + Model: "glm-5.1", Stream: true, OK: true, + TTFB: 500 * time.Millisecond, Latency: 2 * time.Second, + HasUsage: true, PromptTokens: 10, CompletionTokens: 100, TotalTokens: 110, + CacheHitTokens: 40, CacheMissTokens: 60, Credit: 0.02, + }) + c.Record(Delta{ + Model: "glm-5.1", Stream: false, OK: true, + Latency: 1 * time.Second, + HasUsage: true, PromptTokens: 5, CompletionTokens: 50, TotalTokens: 55, + CacheHitTokens: 0, CacheMissTokens: 5, Credit: 0.01, + }) + c.Record(Delta{ + Model: "deepseek-v4.1-flash", Stream: true, OK: false, + TTFB: 300 * time.Millisecond, Latency: 500 * time.Millisecond, + }) + + snap := c.Snapshot() + if len(snap.Models) != 2 { + t.Fatalf("应有 2 个模型, 得到 %d", len(snap.Models)) + } + + glm := snap.Models["glm-5.1"] + if glm == nil { + t.Fatal("glm-5.1 统计缺失") + } + if glm.Requests != 2 || glm.Success != 2 || glm.Streaming != 1 { + t.Errorf("glm-5.1 请求计数错误: %+v", glm) + } + if glm.PromptTokens != 15 || glm.CompletionTokens != 150 || glm.TotalTokens != 165 { + t.Errorf("glm-5.1 token 错误: prompt=%d completion=%d total=%d", + glm.PromptTokens, glm.CompletionTokens, glm.TotalTokens) + } + if glm.CacheHitTokens != 40 || glm.CacheMissTokens != 65 { + t.Errorf("glm-5.1 缓存错误: hit=%d miss=%d", glm.CacheHitTokens, glm.CacheMissTokens) + } + if glm.CreditMilli != 30 { // 0.02 + 0.01 = 0.03 → 30 毫 + t.Errorf("glm-5.1 扣费错误: %d 毫, want 30", glm.CreditMilli) + } + + ds := snap.Models["deepseek-v4.1-flash"] + if ds.Requests != 1 || ds.Failed != 1 { + t.Errorf("失败计数错误: %+v", ds) + } + + // 汇总应为两模型之和。 + if snap.Total.Requests != 3 { + t.Errorf("汇总请求数 = %d, want 3", snap.Total.Requests) + } + if snap.Total.CompletionTokens != 150 { + t.Errorf("汇总输出 token = %d, want 150", snap.Total.CompletionTokens) + } +} + +// TestDeriveComputesRates 派生指标（平均 TTFB / 吞吐 / 命中率）计算正确。 +func TestDeriveComputesRates(t *testing.T) { + m := &ModelStats{ + Model: "glm-5.1", + Requests: 2, Success: 2, + TTFBSumMS: 1000, TTFBCount: 2, // 平均 500ms + LatencySumMS: 3000, LatencyCount: 2, // 平均 1500ms + GenSumMS: 2000, GenCount: 2, // 生成共 2s + PromptTokens: 100, CompletionTokens: 200, TotalTokens: 300, + CacheHitTokens: 300, CacheMissTokens: 100, // 命中率 75% + CreditMilli: 30, // 0.03 + } + d := Derive(m) + + if d.AvgTTFBMS != 500 { + t.Errorf("平均 TTFB = %v, want 500", d.AvgTTFBMS) + } + if d.AvgLatencyMS != 1500 { + t.Errorf("平均延迟 = %v, want 1500", d.AvgLatencyMS) + } + // 吞吐 = 200 token / 2s = 100 tok/s + if d.TokensPerSec != 100 { + t.Errorf("吞吐 = %v tok/s, want 100", d.TokensPerSec) + } + // 命中率 = 300/(300+100) = 0.75 + if d.CacheHitRate != 0.75 { + t.Errorf("命中率 = %v, want 0.75", d.CacheHitRate) + } + if d.Credit != 0.03 { + t.Errorf("扣费 = %v, want 0.03", d.Credit) + } + if d.CreditPerReq != 0.015 { + t.Errorf("每请求扣费 = %v, want 0.015", d.CreditPerReq) + } +} + +// TestDeriveZeroSafe 零值不应产生 NaN/Inf（面板显示会是 "NaN"）。 +func TestDeriveZeroSafe(t *testing.T) { + d := Derive(&ModelStats{Model: "x"}) + if d.AvgTTFBMS != 0 || d.AvgLatencyMS != 0 || d.TokensPerSec != 0 || d.CacheHitRate != 0 || d.CreditPerReq != 0 { + t.Errorf("零值派生应全为 0: %+v", d) + } +} + +// TestDerivedSortsByRequests 模型按请求数降序（面板首屏看热点模型）。 +func TestDerivedSortsByRequests(t *testing.T) { + c := New("") + for i := 0; i < 5; i++ { + c.Record(Delta{Model: "hot", OK: true}) + } + for i := 0; i < 2; i++ { + c.Record(Delta{Model: "cold", OK: true}) + } + c.Record(Delta{Model: "mid", OK: true}) + c.Record(Delta{Model: "mid", OK: true}) + c.Record(Delta{Model: "mid", OK: true}) + + d := c.Derived() + want := []string{"hot", "mid", "cold"} + for i, w := range want { + if d.Models[i].Model != w { + t.Errorf("排序[%d] = %s, want %s", i, d.Models[i].Model, w) + } + } +} + +// TestUnknownModelBucketed 空模型名归入 (unknown)，避免 map 空键。 +func TestUnknownModelBucketed(t *testing.T) { + c := New("") + c.Record(Delta{OK: true}) + snap := c.Snapshot() + if _, ok := snap.Models["(unknown)"]; !ok { + t.Errorf("空模型应归入 (unknown), 得到 %v", snap.Models) + } +} + +// TestPersistence 落盘后重新加载，累计值不丢。 +func TestPersistence(t *testing.T) { + dir := t.TempDir() + fp := filepath.Join(dir, "metrics.json") + + c := New(fp) + // 写足够多让 flushEvery(20) 触发落盘。 + for i := 0; i < 25; i++ { + c.Record(Delta{Model: "glm-5.1", OK: true, HasUsage: true, CompletionTokens: 10}) + } + c.Flush() + + if _, err := os.Stat(fp); err != nil { + t.Fatalf("统计文件应已生成: %v", err) + } + + c2 := New(fp) + snap := c2.Snapshot() + m := snap.Models["glm-5.1"] + if m == nil { + t.Fatal("重启后模型统计丢失") + } + if m.Requests != 25 || m.CompletionTokens != 250 { + t.Errorf("重启后计数错误: requests=%d completion=%d", m.Requests, m.CompletionTokens) + } +} + +// TestPersistenceCorruptFileStartsFresh 文件损坏时从零开始，不 panic。 +func TestPersistenceCorruptFileStartsFresh(t *testing.T) { + dir := t.TempDir() + fp := filepath.Join(dir, "metrics.json") + if err := os.WriteFile(fp, []byte("{not json"), 0o600); err != nil { + t.Fatal(err) + } + c := New(fp) + c.Record(Delta{Model: "m", OK: true}) + if snap := c.Snapshot(); snap.Models["m"].Requests != 1 { + t.Errorf("损坏文件应被忽略并从零开始: %+v", snap.Models) + } +} + +// TestReset 重置清空全部统计。 +func TestReset(t *testing.T) { + c := New("") + c.Record(Delta{Model: "m", OK: true}) + if len(c.Snapshot().Models) != 1 { + t.Fatal("前置条件失败") + } + c.Reset() + if snap := c.Snapshot(); len(snap.Models) != 0 || snap.Total.Requests != 0 { + t.Errorf("重置后应为空: %+v", snap) + } +} + +// TestConcurrentRecord 并发记录不应丢数据或触发竞态（-race 下有完整意义）。 +func TestConcurrentRecord(t *testing.T) { + c := New("") + const n = 200 + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + c.Record(Delta{ + Model: "m", OK: true, HasUsage: true, + CompletionTokens: 1, PromptTokens: 1, TotalTokens: 2, + }) + }(i) + } + wg.Wait() + + m := c.Snapshot().Models["m"] + if m.Requests != n { + t.Errorf("并发丢数据: requests=%d, want %d", m.Requests, n) + } + if m.CompletionTokens != n { + t.Errorf("并发丢 token: %d, want %d", m.CompletionTokens, n) + } +} + +// TestSnapshotIsDeepCopy 快照必须是拷贝：修改快照不应影响内部状态。 +func TestSnapshotIsDeepCopy(t *testing.T) { + c := New("") + c.Record(Delta{Model: "m", OK: true}) + snap := c.Snapshot() + snap.Models["m"].Requests = 999 + if again := c.Snapshot(); again.Models["m"].Requests != 1 { + t.Error("快照未深拷贝：外部修改泄漏进了内部状态") + } +} + +// TestTokensPerSecFallsBackToLatency 无 TTFB 采样（同步请求）时吞吐退回端到端耗时。 +func TestTokensPerSecFallsBackToLatency(t *testing.T) { + m := &ModelStats{ + CompletionTokens: 100, + LatencySumMS: 2000, LatencyCount: 1, // 2s + // GenCount = 0（无流式） + } + d := Derive(m) + if d.TokensPerSec != 50 { // 100/2s + t.Errorf("吞吐 = %v, want 50（应退回端到端耗时）", d.TokensPerSec) + } +} + +// TestCacheNamingVariantsBothCounted 两套缓存命名（OpenAI/Anthropic）都要计入。 +func TestCacheNamingVariantsBothCounted(t *testing.T) { + c := New("") + // OpenAI 风格 + c.Record(Delta{Model: "m", OK: true, HasUsage: true, + CacheHitTokens: 10, CacheMissTokens: 90}) + // Anthropic 风格（cache_read_input_tokens） + c.Record(Delta{Model: "m", OK: true, HasUsage: true, + CacheReadTokens: 50, CacheMissTokens: 50}) + + d := Derive(c.Snapshot().Models["m"]) + // hit = 10 + 50 = 60, miss = 140 → 60/200 = 30% + if d.CacheHitTokens != 60 { + t.Errorf("命中 token = %d, want 60（两套命名都要计）", d.CacheHitTokens) + } + if d.CacheHitRate < 0.299 || d.CacheHitRate > 0.301 { + t.Errorf("命中率 = %v, want 0.3", d.CacheHitRate) + } +} diff --git a/internal/server/handler.go b/internal/server/handler.go index 5825d7c7..80fb9b74 100644 --- a/internal/server/handler.go +++ b/internal/server/handler.go @@ -15,6 +15,7 @@ import ( "workbuddy2api/internal/auth" "workbuddy2api/internal/logfmt" + "workbuddy2api/internal/metrics" "workbuddy2api/internal/pool" "workbuddy2api/internal/prompt" "workbuddy2api/internal/session" @@ -44,6 +45,9 @@ type Config struct { // PromptText custom 模式下注入的系统提示词文本（来自 config.PromptText）。 PromptText string + // Metrics 按模型的请求统计收集器（nil = 未启用，/v1/stats 返回 enabled:false）。 + Metrics *metrics.Collector + // GlobalEnabled global realm 路由开关（config global.enabled，缺省 true）。 // handler 侧第三道闸（与 main 注入 auth 开关、upstream.GlobalEnabled 呼应）： // false（显式逃生门）时即便 auth realm=global 也不提供 global: 模型名 @@ -89,6 +93,8 @@ func NewHandler(cfg Config) *Handler { h := &Handler{cfg: cfg, mux: http.NewServeMux()} h.mux.HandleFunc("POST /v1/chat/completions", h.withAuth(h.chatCompletions)) h.mux.HandleFunc("GET /v1/models", h.withAuth(h.models)) + h.mux.HandleFunc("GET /v1/stats", h.withAuth(h.stats)) + h.mux.HandleFunc("POST /v1/stats/reset", h.withAuth(h.statsReset)) h.mux.HandleFunc("GET /status", h.withAuth(h.status)) h.mux.HandleFunc("GET /healthz", h.healthz) return h @@ -135,6 +141,39 @@ func (h *Handler) healthz(w http.ResponseWriter, r *http.Request) { }) } +// stats 返回按模型聚合的请求统计（面板「统计」页数据源）。 +// +// 采集点放在网关侧而不是面板：网关是所有流量（含绕过面板的客户端）的唯一必经点， +// 只有在这里才能统计到完整调用，且不依赖面板是否在运行。 +func (h *Handler) stats(w http.ResponseWriter, r *http.Request) { + if h.cfg.Metrics == nil { + writeJSON(w, http.StatusOK, map[string]any{ + "enabled": false, + "message": "统计未启用（server.metrics_enabled=false）", + }) + return + } + snap := h.cfg.Metrics.Derived() + writeJSON(w, http.StatusOK, map[string]any{ + "enabled": true, + "since": snap.Since, + "now": snap.Now, + "uptime_sec": snap.UptimeSec, + "total": snap.Total, + "models": snap.Models, + }) +} + +// statsReset 清空统计（运维手动归零，便于观察某个时间点之后的增量）。 +func (h *Handler) statsReset(w http.ResponseWriter, r *http.Request) { + if h.cfg.Metrics == nil { + writeJSON(w, http.StatusOK, map[string]any{"ok": false, "message": "统计未启用"}) + return + } + h.cfg.Metrics.Reset() + writeJSON(w, http.StatusOK, map[string]any{"ok": true, "message": "统计已重置"}) +} + func (h *Handler) status(w http.ResponseWriter, r *http.Request) { total, healthy, cooling, disabled, inFlightFull := h.cfg.Pool.CountsDetailed() sticky := 0 @@ -379,6 +418,7 @@ func (h *Handler) chatCompletions(w http.ResponseWriter, r *http.Request) { // 请求级统计：出口即打一行表格日志（任何路径都会走到）。 st := newChatStat(time.Now(), body, peek.Stream) + st.collector = h.cfg.Metrics defer st.done() tried := map[string]bool{} @@ -600,6 +640,7 @@ func (h *Handler) chatCompletions(w http.ResponseWriter, r *http.Request) { _ = upstream.Stream(w, stats) st.ttfb = stats.TTFB() st.toks, _ = stats.Tokens() + st.usage = stats.Usage() // 成本账本：末帧 usage 带 credit 与 token 总数时记录实测单价， // 供下次选号把免费/便宜的号排在前面。 if credit, ok := stats.Credit(); ok { @@ -623,6 +664,10 @@ func (h *Handler) chatCompletions(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, resp) st.status = http.StatusOK st.toks = completionTokens(resp) + // 统计：非流式聚合响应的 usage 归一化后供采集器取明细。 + if u, ok := resp["usage"].(map[string]any); ok { + st.usage = ParseUsage(u) + } // 成本账本（非流式）：从聚合响应的 usage 取 credit 与 token 总数。 if credit, total, ok := usageCreditTotal(resp); ok { h.cfg.Pool.NoteModelCost(acct.UID, bareModel, credit, total) diff --git a/internal/server/logging.go b/internal/server/logging.go index b931e2ef..e74f5fa8 100644 --- a/internal/server/logging.go +++ b/internal/server/logging.go @@ -10,6 +10,8 @@ import ( "strings" "sync/atomic" "time" + + "workbuddy2api/internal/metrics" ) // chatSeq 进程级请求序号。 @@ -30,6 +32,11 @@ type chatStat struct { status int logged bool + + // collector 统计收集器（nil = 未启用，record() 直接返回）。 + collector *metrics.Collector + // usage 末帧完整 usage（供统计模块取缓存命中/扣费等明细）。 + usage *UsageDetail } // newChatStat 以请求进入 handler 的时刻为起点构造统计对象；toks 默认 -1（usage 缺失）。 @@ -47,9 +54,40 @@ func (s *chatStat) done() { return } s.logged = true + s.record() logChatRow(s.ttfb, time.Since(s.start), s.model, s.mode, s.uid, s.status, s.toks) } +// record 把本次请求写入统计收集器（collector 为 nil 时无操作）。 +// +// 放在 done() 而不是各 return 点：done 是 defer 调用，任何出口（成功/失败/轮转 +// 耗尽/panic 恢复）都会走到，统计不会漏记。 +func (s *chatStat) record() { + if s.collector == nil { + return + } + d := metrics.Delta{ + Model: s.model, + Stream: s.mode == "stream", + OK: s.status >= 200 && s.status < 300, + TTFB: s.ttfb, + Latency: time.Since(s.start), + HasUsage: s.usage != nil, + } + if u := s.usage; u != nil { + d.PromptTokens = u.PromptTokens + d.CompletionTokens = u.CompletionTokens + d.TotalTokens = u.TotalTokens + d.CacheHitTokens = u.CacheHitTokens + d.CacheMissTokens = u.CacheMissTokens + d.CacheWriteTokens = u.CacheWriteTokens + d.CacheReadTokens = u.CacheReadTokens + d.CacheCreationTokens = u.CacheCreationTokens + d.Credit = u.Credit + } + s.collector.Record(d) +} + // chatStatsReader 在流式透传时抓取 SSE 末帧的 usage.completion_tokens 精确值， // 并记录首个 data 帧的 TTFB；原始字节原样返回给下游透传。 // 注意：不做 rune 估算，token 数一律采信上游 usage。 @@ -64,6 +102,9 @@ type chatStatsReader struct { credit float64 // 末帧 usage.credit（本次真实扣费，供成本账本） prompt int // 末帧 usage.prompt_tokens（与 completion 合计折算单价） pend []byte // 已读未返回的行缓存 + + // usage 末帧完整 usage 对象（供统计模块提取缓存命中/扣费等字段）。 + usage *UsageDetail } // newChatStatsReaderSince 以 since 为 TTFB 计时起点（通常是请求进入 handler 的时刻）。 @@ -82,6 +123,9 @@ func (s *chatStatsReader) Tokens() (int, bool) { return s.tokens, s.hasUsage } // 否则收费的号可能被误判 tier0 免费层）。显式 credit:0 仍是合法免费观测（ok=true）。 func (s *chatStatsReader) Credit() (float64, bool) { return s.credit, s.hasUsage && s.hasCredit } +// Usage 返回末帧完整 usage 明细（无 usage 时为 nil）。 +func (s *chatStatsReader) Usage() *UsageDetail { return s.usage } + // TotalTokens 返回本次请求总 token 数（prompt + completion），供成本单价折算。 func (s *chatStatsReader) TotalTokens() int { return s.prompt + s.tokens } @@ -99,22 +143,25 @@ func (s *chatStatsReader) parseSSELine(line string) { s.seen = true s.ttfb = time.Since(s.start) } + // 用宽松 map 解析：usage 字段名在不同模型/区域间有差异（见 UsageDetail 注释）， + // 结构体标签写死会漏字段。 var chunk struct { - Usage *struct { - CompletionTokens int `json:"completion_tokens"` - PromptTokens int `json:"prompt_tokens"` - Credit *float64 `json:"credit"` // 指针区分「缺失」与「显式 0」 - } `json:"usage"` + Usage map[string]any `json:"usage"` } if json.Unmarshal([]byte(payload), &chunk) != nil || chunk.Usage == nil { return } s.hasUsage = true - s.tokens = chunk.Usage.CompletionTokens - s.prompt = chunk.Usage.PromptTokens - if chunk.Usage.Credit != nil { + u := ParseUsage(chunk.Usage) + s.usage = u + s.tokens = int(u.CompletionTokens) + s.prompt = int(u.PromptTokens) + // hasCredit 语义：仅当 credit 字段显式出现**且非 null**（含显式 0）才置位。 + // 缺失≠0 —— 不能把「缺观测」当「0 成本」写入账本；credit:null 同样视为缺失 + //（上游用 null 表达"本次无扣费观测"，与显式 0 语义不同）。 + if v, ok := chunk.Usage["credit"]; ok && v != nil { s.hasCredit = true - s.credit = *chunk.Usage.Credit + s.credit = u.Credit } } @@ -176,6 +223,82 @@ func usageCreditTotal(resp map[string]any) (credit float64, total int, ok bool) return c, int(pt) + int(ct), true } +// UsageDetail 上游 usage 的归一化视图。 +// +// 为什么用宽松 map 而非结构体标签：usage 字段名在不同模型/区域间有差异 +// （OpenAI 风格的 prompt_cache_hit_tokens 与 Anthropic 风格的 +// cache_read_input_tokens 都出现过），写死标签会漏字段。 +// 缺失字段一律按 0 处理（不猜、不估算）。 +type UsageDetail struct { + PromptTokens int64 + CompletionTokens int64 + TotalTokens int64 + + CacheHitTokens int64 + CacheMissTokens int64 + CacheWriteTokens int64 + CacheReadTokens int64 + CacheCreationTokens int64 + + // Credit 上游返回的实际扣费（部分模型带 credit 字段）。 + Credit float64 +} + +// ParseUsage 把上游 usage 对象归一化为 UsageDetail。 +func ParseUsage(u map[string]any) *UsageDetail { + d := &UsageDetail{ + PromptTokens: int64Field(u, "prompt_tokens"), + CompletionTokens: int64Field(u, "completion_tokens"), + TotalTokens: int64Field(u, "total_tokens"), + + // OpenAI 风格缓存字段 + CacheHitTokens: int64Field(u, "prompt_cache_hit_tokens"), + CacheMissTokens: int64Field(u, "prompt_cache_miss_tokens"), + CacheWriteTokens: int64Field(u, "prompt_cache_write_tokens"), + // Anthropic 风格缓存字段 + CacheReadTokens: int64Field(u, "cache_read_input_tokens"), + CacheCreationTokens: int64Field(u, "cache_creation_input_tokens"), + + Credit: float64Field(u, "credit"), + } + // 两套命名都有时取较大值，避免重复累计导致命中率虚高 + // （同一份数据被两个字段名各报一次的场景）。 + if v := int64Field(u, "cached_tokens"); v > d.CacheHitTokens { + d.CacheHitTokens = v + } + // total 缺失时用 prompt+completion 补齐（便于面板直接展示）。 + if d.TotalTokens == 0 { + d.TotalTokens = d.PromptTokens + d.CompletionTokens + } + return d +} + +// int64Field 从 map 取整数字段（兼容 JSON number 的 float64 与字符串数字）。 +func int64Field(m map[string]any, key string) int64 { + switch v := m[key].(type) { + case float64: + return int64(v) + case int64: + return v + case int: + return int64(v) + } + return 0 +} + +// float64Field 从 map 取浮点字段。 +func float64Field(m map[string]any, key string) float64 { + switch v := m[key].(type) { + case float64: + return v + case int64: + return float64(v) + case int: + return float64(v) + } + return 0 +} + // uidPrefix 只显示 uid 前 8 位；空 uid 显示 "-"。 func uidPrefix(uid string) string { if uid == "" {