From 82268451122133dfcf02418de8625f2a78432c42 Mon Sep 17 00:00:00 2001 From: zhanghao Date: Tue, 10 Nov 2026 17:33:07 +0800 Subject: [PATCH] bodylog: export actual_usage's cached_tokens as its own counter bodylog_cached_tokens_total counts usage.prompt_tokens_details.cached_tokens -- what clients and billing see. Some sglang builds also return actual_usage with the engine's own count, which can differ (e.g. 2048 vs 3072), and that number reached no metric. listener: the details line gets actual_cached_tokens, from resp_meta.actual_usage.prompt_tokens_details.cached_tokens, only when that is a number, so "not reported" and "reported 0" stay apart; the DuckDB details schema gains the column (older files read it as NULL). exporter: bodylog_actual_cached_tokens_total, same labels as bodylog_cached_tokens_total, fed only by lines that carry the field. --- bodylog-exporter-go/metrics.go | 29 +++++++---- .../metrics_actual_cached_test.go | 48 +++++++++++++++++++ bodylog-listener-go/main.go | 19 +++++++- .../resp_meta_actual_usage_test.go | 38 +++++++++++++++ 4 files changed, 124 insertions(+), 10 deletions(-) create mode 100644 bodylog-exporter-go/metrics_actual_cached_test.go diff --git a/bodylog-exporter-go/metrics.go b/bodylog-exporter-go/metrics.go index 0437ad5..745eba1 100644 --- a/bodylog-exporter-go/metrics.go +++ b/bodylog-exporter-go/metrics.go @@ -28,8 +28,11 @@ type detailRecord struct { PromptTokens int64 `json:"prompt_tokens"` CompletionTokens int64 `json:"completion_tokens"` CachedTokens int64 `json:"cached_tokens"` - TotalTokens int64 `json:"total_tokens"` - ReasoningTokens int64 `json:"reasoning_tokens"` + // actual_usage's cached_tokens (some sglang builds report a second count next to usage); + // absent from the line when the response had none + ActualCachedTokens *int64 `json:"actual_cached_tokens"` + TotalTokens int64 `json:"total_tokens"` + ReasoningTokens int64 `json:"reasoning_tokens"` } type metrics struct { @@ -37,6 +40,7 @@ type metrics struct { promptTok *prometheus.CounterVec // {backend,model} completionTok *prometheus.CounterVec cachedTok *prometheus.CounterVec + actualCachedTok *prometheus.CounterVec reasoningTok *prometheus.CounterVec totalTok *prometheus.CounterVec reqBytes *prometheus.CounterVec @@ -84,12 +88,16 @@ func newMetrics(reg *prometheus.Registry) *metrics { promptTok: ctr("bodylog_prompt_tokens_total", "prompt token 累计", srbm), completionTok: ctr("bodylog_completion_tokens_total", "completion token 累计", srbm), cachedTok: ctr("bodylog_cached_tokens_total", "cached token 累计", srbm), - reasoningTok: ctr("bodylog_reasoning_tokens_total", "reasoning token 累计", srbm), - totalTok: ctr("bodylog_total_tokens_total", "total token 累计", srbm), - reqBytes: ctr("bodylog_req_bytes_total", "请求体字节累计", srbm), - respBytes: ctr("bodylog_resp_bytes_total", "响应体字节累计", srbm), - finishReason: ctr("bodylog_finish_reason_total", "按 finish_reason 计数", []string{"service", "route", "backend", "model", "finish_reason"}), - rt: hist("bodylog_rt_seconds", "总响应时间(秒)", []string{"service", "route", "backend", "model", "stream"}), + // usage.prompt_tokens_details.cached_tokens is what clients and billing see; some sglang + // builds also send actual_usage with the engine's own count, which can differ. This is that + // one, counted only for responses that reported it. + actualCachedTok: ctr("bodylog_actual_cached_tokens_total", "actual_usage.prompt_tokens_details.cached_tokens 累计(仅上游返回 actual_usage 的请求)", srbm), + reasoningTok: ctr("bodylog_reasoning_tokens_total", "reasoning token 累计", srbm), + totalTok: ctr("bodylog_total_tokens_total", "total token 累计", srbm), + reqBytes: ctr("bodylog_req_bytes_total", "请求体字节累计", srbm), + respBytes: ctr("bodylog_resp_bytes_total", "响应体字节累计", srbm), + finishReason: ctr("bodylog_finish_reason_total", "按 finish_reason 计数", []string{"service", "route", "backend", "model", "finish_reason"}), + rt: hist("bodylog_rt_seconds", "总响应时间(秒)", []string{"service", "route", "backend", "model", "stream"}), // prompt_bucket:输入长度分档,用于“某 context 区间的 TTFT 分位数”。见 promptBucket()。 ttft: hist("bodylog_ttft_seconds", "首 token 时间/TTFT(秒,仅流式)", srbmp), // 2026-08-26:分母从 rt 改为 rt-frt(扣除 prefill),仅流式 + 下限过滤,与 openresty 引擎 @@ -106,7 +114,7 @@ func newMetrics(reg *prometheus.Registry) *metrics { recovers: prometheus.NewCounter(prometheus.CounterOpts{Name: "bodylog_exporter_recovery_total", Help: "跨天缺口 HTTP 补读次数"}), } reg.MustRegister( - m.requests, m.promptTok, m.completionTok, m.cachedTok, m.reasoningTok, m.totalTok, + m.requests, m.promptTok, m.completionTok, m.cachedTok, m.actualCachedTok, m.reasoningTok, m.totalTok, m.reqBytes, m.respBytes, m.finishReason, m.rt, m.ttft, m.outTokPerSec, m.overallTokPerSec, m.decodeTokPerSec, m.lines, m.offset, m.lastTs, m.recovers, @@ -173,6 +181,9 @@ func (m *metrics) observe(d detailRecord) { if d.CachedTokens > 0 { m.cachedTok.WithLabelValues(service, route, backend, model).Add(float64(d.CachedTokens)) } + if d.ActualCachedTokens != nil && *d.ActualCachedTokens > 0 { + m.actualCachedTok.WithLabelValues(service, route, backend, model).Add(float64(*d.ActualCachedTokens)) + } if d.ReasoningTokens > 0 { m.reasoningTok.WithLabelValues(service, route, backend, model).Add(float64(d.ReasoningTokens)) } diff --git a/bodylog-exporter-go/metrics_actual_cached_test.go b/bodylog-exporter-go/metrics_actual_cached_test.go new file mode 100644 index 0000000..3161771 --- /dev/null +++ b/bodylog-exporter-go/metrics_actual_cached_test.go @@ -0,0 +1,48 @@ +package main + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus" +) + +// counterTotal sums every series of one counter family. +func counterTotal(t *testing.T, reg *prometheus.Registry, name string) float64 { + t.Helper() + fams, err := reg.Gather() + if err != nil { + t.Fatal(err) + } + var sum float64 + for _, f := range fams { + if f.GetName() == name { + for _, m := range f.GetMetric() { + sum += m.GetCounter().GetValue() + } + } + } + return sum +} + +// Some sglang builds report a second token count, actual_usage, next to usage; the listener +// writes its cached_tokens to the details line as actual_cached_tokens. It gets its own counter: +// bodylog_cached_tokens_total keeps counting usage's figure (what clients and billing see). +func TestActualCachedTokensCounter(t *testing.T) { + reg := prometheus.NewRegistry() + tl := &tailer{m: newMetrics(reg)} + line := func(extra string) string { + return `{"ts":"2026-11-10T16:00:00+08:00","ts_end":"2026-11-10T16:00:01+08:00","request_id":"r",` + + `"stream":true,"status":200,"backend":"10.0.0.4:8050","model":"kimi-k3","finish_reason":"stop",` + + `"frt":0.5,"lct":1.0,"rt":1.5,"prompt_tokens":4096,"completion_tokens":10,"cached_tokens":2048` + extra + `}` + "\n" + } + tl.handleLine([]byte(line(`,"actual_cached_tokens":3072`))) + tl.handleLine([]byte(line(`,"actual_cached_tokens":0`))) // reported, nothing cached + tl.handleLine([]byte(line(``))) // a backend without actual_usage + + if got := counterTotal(t, reg, "bodylog_cached_tokens_total"); got != 3*2048 { + t.Errorf("bodylog_cached_tokens_total = %v, want %v (usage's figure, every request)", got, 3*2048) + } + if got := counterTotal(t, reg, "bodylog_actual_cached_tokens_total"); got != 3072 { + t.Errorf("bodylog_actual_cached_tokens_total = %v, want 3072 (only what actual_usage reported)", got) + } +} diff --git a/bodylog-listener-go/main.go b/bodylog-listener-go/main.go index ec3c15a..c05434d 100644 --- a/bodylog-listener-go/main.go +++ b/bodylog-listener-go/main.go @@ -171,6 +171,10 @@ type detailRecord struct { CachedTokens int64 `json:"cached_tokens"` TotalTokens int64 `json:"total_tokens"` ReasoningTokens int64 `json:"reasoning_tokens"` + // actual_usage.prompt_tokens_details.cached_tokens (some sglang builds send a second token + // count next to usage). Absent unless the response had actual_usage, so "not reported" and + // "reported 0" stay apart. + ActualCachedTokens *int64 `json:"actual_cached_tokens,omitempty"` } // extractDetail: 从 assembleEntry 解析好的 map 里抽出 metrics 明细，不带任何正文。 @@ -203,6 +207,10 @@ func extractDetail(m map[string]any) detailRecord { TotalTokens: u.total, ReasoningTokens: u.reasoning, } + if u.hasActualCached { + v := u.actualCached + d.ActualCachedTokens = &v + } if rm, ok := m["resp_meta"].(map[string]any); ok { d.Model = getString(rm["model"]) d.FinishReason = getString(rm["finish_reason"]) @@ -1188,6 +1196,8 @@ func normalizePeer(m map[string]any) string { type usageTokens struct { prompt, cached, compl, total, reasoning int64 + actualCached int64 // actual_usage.prompt_tokens_details.cached_tokens + hasActualCached bool // the response carried actual_usage with that field } // extractUsage: 从 resp_meta.usage 抽 token 计数，兼容 OpenAI(prompt/completion_tokens) @@ -1200,6 +1210,13 @@ func extractUsage(m map[string]any) usageTokens { if !ok { return u } + if au, ok := rm["actual_usage"].(map[string]any); ok { + if d, ok := au["prompt_tokens_details"].(map[string]any); ok { + if v, ok := d["cached_tokens"].(float64); ok { // a number; null or missing = not reported + u.actualCached, u.hasActualCached = int64(v), true + } + } + } us, ok := rm["usage"].(map[string]any) if !ok { return u @@ -1317,7 +1334,7 @@ const detailsColumns = "{" + "'finish_reason':'VARCHAR','frt':'DOUBLE','lct':'DOUBLE','rt':'DOUBLE'," + "'chunk_count':'BIGINT','req_bytes':'BIGINT','resp_bytes':'BIGINT'," + "'prompt_tokens':'BIGINT','completion_tokens':'BIGINT','cached_tokens':'BIGINT'," + - "'total_tokens':'BIGINT','reasoning_tokens':'BIGINT'}" + "'total_tokens':'BIGINT','reasoning_tokens':'BIGINT','actual_cached_tokens':'BIGINT'}" // readDetailsJSON: 用显式 schema 读明细 jsonl 的 DuckDB 表达式。arg 是单个 sqlStr 路径 // 或文件名列表（read_json 两者都接受）。rotateDetails 与 metricsHandler 共用，保证写/读 diff --git a/bodylog-listener-go/resp_meta_actual_usage_test.go b/bodylog-listener-go/resp_meta_actual_usage_test.go index f5d52fa..a488ebd 100644 --- a/bodylog-listener-go/resp_meta_actual_usage_test.go +++ b/bodylog-listener-go/resp_meta_actual_usage_test.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "strings" "testing" ) @@ -58,3 +59,40 @@ func TestRespMetaKeepsActualUsage(t *testing.T) { t.Errorf("actual_usage present though upstream sent none: %v", meta) } } + +// The details line (what bodylog-exporter reads) carries actual_usage's cached_tokens as +// actual_cached_tokens next to cached_tokens -- only when the response reported one. +func TestDetailActualCachedTokens(t *testing.T) { + usage := `{"prompt_tokens":4096,"completion_tokens":128,"total_tokens":4224,"prompt_tokens_details":{"cached_tokens":2048}}` + cases := []struct { + name, extra string + want *int64 + }{ + {"reported", `,"actual_usage":{"prompt_tokens":4096,"prompt_tokens_details":{"cached_tokens":3072}}`, ptr(3072)}, + {"reported zero", `,"actual_usage":{"prompt_tokens_details":{"cached_tokens":0}}`, ptr(0)}, + {"null", `,"actual_usage":{"prompt_tokens_details":{"cached_tokens":null}}`, nil}, + {"no details", `,"actual_usage":{"prompt_tokens":4096}`, nil}, + {"absent", ``, nil}, + } + for _, c := range cases { + body := `{"choices":[{"message":{"content":"hi"},"finish_reason":"stop"}],"usage":` + usage + c.extra + `}` + m := map[string]any{"status": float64(200), "resp_meta": buildRespMeta(extractResp(body))} + d := extractDetail(m) + if d.CachedTokens != 2048 { + t.Errorf("%s: cached_tokens = %d, want 2048 (usage is unaffected)", c.name, d.CachedTokens) + } + switch { + case c.want == nil && d.ActualCachedTokens != nil: + t.Errorf("%s: actual_cached_tokens = %d, want absent", c.name, *d.ActualCachedTokens) + case c.want != nil && (d.ActualCachedTokens == nil || *d.ActualCachedTokens != *c.want): + t.Errorf("%s: actual_cached_tokens = %v, want %d", c.name, d.ActualCachedTokens, *c.want) + } + line, _ := json.Marshal(d) + has := strings.Contains(string(line), `"actual_cached_tokens"`) + if has != (c.want != nil) { + t.Errorf("%s: details line has actual_cached_tokens=%v, want %v: %s", c.name, has, c.want != nil, line) + } + } +} + +func ptr(v int64) *int64 { return &v }