mirror of
https://github.com/lukaszraczylo/traefik-llmgateway.git
synced 2026-10-15 05:29:28 +00:00
merge: expose token-accounting provenance
This commit is contained in:
@@ -303,9 +303,20 @@ type adminProviderView struct {
|
||||
// declared after the scalar block as costing extra GC pointer-scan
|
||||
// bytes. See timeout.go's watchdogBody for the identical convention.
|
||||
Latency map[string]adminLatencyView `json:"latency,omitempty"`
|
||||
Type string `json:"type"`
|
||||
BaseURL string `json:"baseUrl"`
|
||||
LastErr string `json:"lastErr,omitempty"`
|
||||
// Provenance summarizes non-reported usage accounting for this
|
||||
// provider (feat: expose token-accounting provenance) — keyed by
|
||||
// provenance kind, "estimated" or "unbilled" only (see
|
||||
// adminProvenanceView's own doc comment for why "reported" is never a
|
||||
// key here). Sourced from the SAME in-process g.provenance
|
||||
// accumulator /metrics reads (metrics.go). nil (omitted) for a
|
||||
// provider with no estimated/unbilled outcome yet, or for a
|
||||
// deployment with metrics collection gated off entirely — mirroring
|
||||
// Latency's identical "nil means no observations yet" convention
|
||||
// immediately above.
|
||||
Provenance map[string]adminProvenanceView `json:"provenance,omitempty"`
|
||||
Type string `json:"type"`
|
||||
BaseURL string `json:"baseUrl"`
|
||||
LastErr string `json:"lastErr,omitempty"`
|
||||
// HealthState is this provider's discovery circuit breaker state
|
||||
// (feat/provider-health): "closed" (normal), "open" (backing off
|
||||
// after repeated discovery failures — maybeRefresh skips it until
|
||||
@@ -417,6 +428,48 @@ func buildAdminLatencyViews(snaps []latencySnapshot) map[string]map[string]admin
|
||||
return out
|
||||
}
|
||||
|
||||
// adminProvenanceView is one provider's one non-reported provenance
|
||||
// kind's compact usage-accounting summary (feat: expose token-accounting
|
||||
// provenance) — adminProviderView.Provenance's value type, keyed by
|
||||
// provenance kind ("estimated" or "unbilled" — provenanceEstimated/
|
||||
// provenanceUnbilled, metrics.go) in adminProviderView.Provenance's outer
|
||||
// map. "reported" is deliberately never a key here: it is the default,
|
||||
// healthy case AttemptsDay/FailuresDay above already partially describe,
|
||||
// and duplicating its own count would not answer any operator question
|
||||
// this compact surface (buildAdminLatencyViews' own "compact, no
|
||||
// per-model breakdown" ruling, GET /admin/api/overview's ~113KB/5s-poll
|
||||
// budget) is not already better placed to answer — the full reported/
|
||||
// estimated/unbilled breakdown, by request AND by token, lives on
|
||||
// /metrics (llmgateway_usage_provenance_requests_total/-_tokens_total)
|
||||
// for an operator who wants the precise fraction.
|
||||
type adminProvenanceView struct {
|
||||
Requests int64 `json:"requests"`
|
||||
Tokens int64 `json:"tokens"`
|
||||
}
|
||||
|
||||
// buildAdminProvenanceViews groups snaps (g.provenance.snapshot(),
|
||||
// metrics.go) by provider, discarding the "reported" provenance entirely
|
||||
// (adminProvenanceView's own doc comment) — mirroring
|
||||
// buildAdminLatencyViews' identical "compact, provider-keyed outer map"
|
||||
// shape immediately above. A provider with no estimated/unbilled
|
||||
// observations at all is simply absent from the outer map, so
|
||||
// adminOverviewResponse's own lookup (buildAdminOverview, below) yields
|
||||
// nil for it — adminProviderView.Provenance's own documented "nil means
|
||||
// no estimated/unbilled accounting yet".
|
||||
func buildAdminProvenanceViews(snaps []provenanceSnapshot) map[string]map[string]adminProvenanceView {
|
||||
out := make(map[string]map[string]adminProvenanceView)
|
||||
for _, s := range snaps {
|
||||
if s.key.provenance == provenanceReported {
|
||||
continue
|
||||
}
|
||||
if out[s.key.provider] == nil {
|
||||
out[s.key.provider] = make(map[string]adminProvenanceView)
|
||||
}
|
||||
out[s.key.provider][s.key.provenance] = adminProvenanceView{Requests: s.requests, Tokens: s.tokens}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// adminModelMetaView is one upstream model's resolved metadata (feature
|
||||
// v0.23: context window, per-token cost) within its provider —
|
||||
// adminProviderView.ModelMeta's value type. Every field is a pointer,
|
||||
@@ -559,6 +612,10 @@ func (g *Gateway) buildAdminOverview() adminOverviewResponse {
|
||||
// for the whole response, mirroring providerUsage's own one-batched-
|
||||
// read-for-every-provider shape above, not a per-provider read.
|
||||
latencyViews := buildAdminLatencyViews(g.latency.snapshot())
|
||||
// feat: expose token-accounting provenance — one g.provenance.
|
||||
// snapshot() call for the whole response, mirroring g.latency.
|
||||
// snapshot()'s own one-batched-read-for-every-provider shape above.
|
||||
provenanceViews := buildAdminProvenanceViews(g.provenance.snapshot())
|
||||
|
||||
providers := make([]adminProviderView, len(snaps))
|
||||
mi := 0
|
||||
@@ -593,6 +650,7 @@ func (g *Gateway) buildAdminOverview() adminOverviewResponse {
|
||||
ModelRates: modelRates,
|
||||
ModelMeta: modelMeta,
|
||||
Latency: latencyViews[s.name],
|
||||
Provenance: provenanceViews[s.name],
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+10
-10
File diff suppressed because one or more lines are too long
@@ -631,6 +631,78 @@ func TestAdminOverview_LatencySummary_AvgReflectsSeededObservations(t *testing.T
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdminOverview_ProvenanceSummary_ExcludesReportedIncludesEstimatedAndUnbilled
|
||||
// proves buildAdminProvenanceViews' own filtering (admin.go, feat:
|
||||
// expose token-accounting provenance): "reported" observations never
|
||||
// surface on this compact admin summary (the operator already sees
|
||||
// AttemptsDay/FailuresDay for the healthy default), while "estimated"
|
||||
// and "unbilled" both do, as DISTINCT keys — never merged.
|
||||
//
|
||||
// MUTATION VERIFIED: removing the `if s.key.provenance ==
|
||||
// provenanceReported { continue }` guard from buildAdminProvenanceViews
|
||||
// (admin.go) made this test fail — zeta.Provenance gained a leaked
|
||||
// "reported" entry (Requests=1, Tokens=15) that must never appear here.
|
||||
// Reverted before committing.
|
||||
func TestAdminOverview_ProvenanceSummary_ExcludesReportedIncludesEstimatedAndUnbilled(t *testing.T) {
|
||||
t.Parallel()
|
||||
cfg := newAdminTestConfig()
|
||||
h, gw := newAdminGatewayHandle(t, cfg)
|
||||
|
||||
// zeta: one reported observation (must never surface below) and one
|
||||
// estimated observation.
|
||||
gw.provenance.record("zeta", provenanceReported, 15)
|
||||
gw.provenance.record("zeta", provenanceEstimated, 7)
|
||||
|
||||
// alpha: two unbilled observations, seeded separately to prove
|
||||
// requests accumulates across calls rather than overwriting.
|
||||
gw.provenance.record("alpha", provenanceUnbilled, 0)
|
||||
gw.provenance.record("alpha", provenanceUnbilled, 0)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, adminRequest(http.MethodGet, adminOverviewPath, "sk-admin1"))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200, body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
var got adminOverviewResponse
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
|
||||
t.Fatalf("decode: %v", err)
|
||||
}
|
||||
var zeta, alpha *adminProviderView
|
||||
for i := range got.Providers {
|
||||
switch got.Providers[i].Name {
|
||||
case "zeta":
|
||||
zeta = &got.Providers[i]
|
||||
case "alpha":
|
||||
alpha = &got.Providers[i]
|
||||
}
|
||||
}
|
||||
if zeta == nil || alpha == nil {
|
||||
t.Fatalf("providers = %+v, want entries named zeta and alpha", got.Providers)
|
||||
}
|
||||
|
||||
if _, ok := zeta.Provenance[provenanceReported]; ok {
|
||||
t.Errorf(`zeta.Provenance has a %q entry: %+v — "reported" must never surface on this compact admin summary`, provenanceReported, zeta.Provenance)
|
||||
}
|
||||
est, ok := zeta.Provenance[provenanceEstimated]
|
||||
if !ok {
|
||||
t.Fatalf("zeta.Provenance missing %q entry: %+v", provenanceEstimated, zeta.Provenance)
|
||||
}
|
||||
if est.Requests != 1 || est.Tokens != 7 {
|
||||
t.Errorf("zeta estimated = %+v, want {Requests:1 Tokens:7}", est)
|
||||
}
|
||||
|
||||
unb, ok := alpha.Provenance[provenanceUnbilled]
|
||||
if !ok {
|
||||
t.Fatalf("alpha.Provenance missing %q entry: %+v", provenanceUnbilled, alpha.Provenance)
|
||||
}
|
||||
if unb.Requests != 2 || unb.Tokens != 0 {
|
||||
t.Errorf("alpha unbilled = %+v, want {Requests:2 Tokens:0}", unb)
|
||||
}
|
||||
if _, ok := alpha.Provenance[provenanceEstimated]; ok {
|
||||
t.Errorf(`alpha.Provenance has a %q entry despite only unbilled observations being seeded: %+v — estimated and unbilled must never conflate`, provenanceEstimated, alpha.Provenance)
|
||||
}
|
||||
}
|
||||
|
||||
// --- usage: math against seeded counters ---
|
||||
|
||||
func TestAdminUsage_MathAgainstSeededCounters(t *testing.T) {
|
||||
|
||||
+15
-1
@@ -597,7 +597,17 @@ type Gateway struct {
|
||||
// identical "safe even off a bare &Gateway{} literal" convention
|
||||
// above.
|
||||
latency *latencyStore
|
||||
name string
|
||||
// provenance is the in-process usage-accounting-provenance accumulator
|
||||
// (feat: expose token-accounting provenance, metrics.go's
|
||||
// provenanceStore) — recordUsageProvenance's only write target, and
|
||||
// writeProvenanceMetrics/buildAdminProvenanceViews' (admin.go) own
|
||||
// read source. Always constructed, here, by newGateway, even when
|
||||
// Config.Metrics is nil or disabled, mirroring g.latency's own
|
||||
// identical "always-present, usually-empty, nil-receiver-safe"
|
||||
// convention immediately above — see that field's own doc comment for
|
||||
// the full reasoning, which applies here unchanged.
|
||||
provenance *provenanceStore
|
||||
name string
|
||||
// failoverLogGate rate-limits runMeteredCall's generic "failing over"
|
||||
// log line (routes_unified.go) to once per storeErrorLogEvery
|
||||
// (adversarial-review fix, F10) — reuses auth.go's own logGate type,
|
||||
@@ -744,6 +754,10 @@ func newGateway(ctx context.Context, next http.Handler, config *Config, name str
|
||||
// comment above for why an always-present, usually-empty store is the
|
||||
// right default rather than a conditional nil.
|
||||
g.latency = newLatencyStore()
|
||||
// feat: expose token-accounting provenance — always constructed, here,
|
||||
// regardless of whether Config.Metrics is nil/disabled, mirroring
|
||||
// g.latency's own identical "always-present" ruling immediately above.
|
||||
g.provenance = newProvenanceStore()
|
||||
|
||||
// A malformed Config.Metrics.AllowedCIDRs entry is a constructor
|
||||
// error (validate-at-construction), not a silently-ignored allowlist
|
||||
|
||||
+166
-4
@@ -431,10 +431,13 @@ func (m *metricWriter) histogram(name string, labels []metricLabel, buckets []in
|
||||
const aggregationNoteStoreBacked = " AGGREGATION ACROSS REPLICAS: with Redis configured, this value is the gateway-wide total read from shared limit state and is IDENTICAL on every replica's own /metrics response — aggregate with max(), never sum(), or a multi-replica deployment overcounts by the replica count. Without Redis (the in-memory fallback), this SAME family becomes genuinely per-process instead, and sum() becomes the correct aggregation. Know which one you are running."
|
||||
|
||||
// aggregationNotePerProcess is appended to llmgateway_rate_limit_
|
||||
// rejections_total's HELP text: limiter.rejections (limits.go) is a
|
||||
// bare in-process map, never written to or read from Redis regardless
|
||||
// of Config.Redis — unlike every store-backed family above, this one
|
||||
// never flips, and sum() is always the right cross-replica aggregation.
|
||||
// rejections_total's HELP text, and to both usage-provenance counter
|
||||
// families' HELP text below (llmgateway_usage_provenance_requests_total/
|
||||
// -_tokens_total, feat: expose token-accounting provenance): each is a
|
||||
// bare in-process store — limiter.rejections and g.provenance
|
||||
// respectively — never written to or read from Redis regardless of
|
||||
// Config.Redis. Unlike every store-backed family above, none of these
|
||||
// ever flips, and sum() is always the right cross-replica aggregation.
|
||||
const aggregationNotePerProcess = " AGGREGATION ACROSS REPLICAS: this counter lives only in this process's memory, never in Redis, regardless of Config.Redis — aggregate with sum(), which is always correct for it, unlike the store-backed families above."
|
||||
|
||||
// aggregationNoteProviderHealthy is appended to llmgateway_provider_
|
||||
@@ -468,6 +471,7 @@ func (g *Gateway) renderMetrics() []byte {
|
||||
g.writeUsageMetrics(&m)
|
||||
g.writeProviderMetrics(&m)
|
||||
g.writeLatencyMetrics(&m)
|
||||
g.writeProvenanceMetrics(&m)
|
||||
g.writeRejectionMetrics(&m)
|
||||
g.writeStoreHealthMetrics(&m)
|
||||
return m.buf.Bytes()
|
||||
@@ -961,6 +965,164 @@ func (g *Gateway) writeLatencyMetrics(m *metricWriter) {
|
||||
}
|
||||
}
|
||||
|
||||
// --- usage provenance (feat: expose token-accounting provenance) ---
|
||||
//
|
||||
// Token accounting has three provenances (usage.estimated's own doc
|
||||
// comment, limits.go), and before this section nothing on any read
|
||||
// surface distinguished them: REPORTED (the provider returned real
|
||||
// usage — the normal, accurate case), ESTIMATED (a non-streaming
|
||||
// response carried none, so prompt is substituted with
|
||||
// ceil(len(body)/4) and completion is billed as zero — runUnified's own
|
||||
// recording call site, routes_unified.go), and UNBILLED (a streaming
|
||||
// response carried none, so the request is counted but zero tokens are
|
||||
// billed). This section adds the metric surface only; billing itself is
|
||||
// unchanged — see runUnified's own doc comment at its recording call
|
||||
// site for the guarantee that no value recorded here ever feeds back
|
||||
// into what gets billed.
|
||||
//
|
||||
// ESTIMATED and UNBILLED are deliberately never merged into one "not
|
||||
// reported" signal: estimated substitutes an approximation (a nonzero
|
||||
// prompt count is still billed), unbilled charges nothing at all. A
|
||||
// provider silently ignoring stream_options.include_usage serves
|
||||
// completions entirely free against every configured budget, and that
|
||||
// is the one fact this feature exists to make visible — collapsing the
|
||||
// two would hide it again.
|
||||
|
||||
const (
|
||||
provenanceReported = "reported"
|
||||
provenanceEstimated = "estimated"
|
||||
provenanceUnbilled = "unbilled"
|
||||
)
|
||||
|
||||
// provenanceKey identifies one (provider, provenance) series pair
|
||||
// provenanceStore accumulates under. provider is always one of this
|
||||
// deployment's configured provider names (config-bounded, never user
|
||||
// input, exactly like latencyKey.provider above), and provenance is
|
||||
// always one of the three package-level constants above — the total key
|
||||
// space is bounded by (configured providers) x 3, so no rotation/cap
|
||||
// logic is needed here, unlike rejectionCounter's own user/group-name
|
||||
// keys (limits.go), which ARE operator/end-user controlled and can churn
|
||||
// without bound.
|
||||
type provenanceKey struct {
|
||||
provider string
|
||||
provenance string
|
||||
}
|
||||
|
||||
// provenanceCounts is one provenanceKey's accumulated totals: requests
|
||||
// (how many accounted candidate outcomes were recorded under this
|
||||
// provenance) and tokens (their summed result.total() — the SAME number
|
||||
// runUnified already passed to accounting, never a second count),
|
||||
// answering the task brief's "what fraction of billed tokens came from
|
||||
// real reported usage" question directly: sum(tokens where provenance ==
|
||||
// reported) / sum(tokens across every provenance).
|
||||
type provenanceCounts struct {
|
||||
requests int64
|
||||
tokens int64
|
||||
}
|
||||
|
||||
// provenanceStore is the Gateway's in-process, per-replica usage-
|
||||
// provenance accumulator — mirrors latencyStore's own shape and
|
||||
// nil-receiver-safe convention above for the identical reason: a Gateway
|
||||
// assembled directly as a bare &Gateway{} literal (bypassing newGateway,
|
||||
// as a few older tests do) degrades to "no provenance observed" rather
|
||||
// than a nil-pointer panic.
|
||||
type provenanceStore struct {
|
||||
data map[provenanceKey]*provenanceCounts
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// newProvenanceStore returns an empty, ready-to-use provenanceStore.
|
||||
func newProvenanceStore() *provenanceStore {
|
||||
return &provenanceStore{data: make(map[provenanceKey]*provenanceCounts)}
|
||||
}
|
||||
|
||||
// record accumulates one candidate outcome under (provider, provenance):
|
||||
// one request, plus tokens — allocating a fresh provenanceCounts the
|
||||
// first time key is seen.
|
||||
func (s *provenanceStore) record(provider, provenance string, tokens int64) {
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
key := provenanceKey{provider: provider, provenance: provenance}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
c := s.data[key]
|
||||
if c == nil {
|
||||
c = &provenanceCounts{}
|
||||
s.data[key] = c
|
||||
}
|
||||
c.requests++
|
||||
c.tokens += tokens
|
||||
}
|
||||
|
||||
// provenanceSnapshot is one provenanceKey's fully-copied counts, safe to
|
||||
// read after provenanceStore.snapshot returns without holding its lock.
|
||||
type provenanceSnapshot struct {
|
||||
key provenanceKey
|
||||
requests int64
|
||||
tokens int64
|
||||
}
|
||||
|
||||
// snapshot returns a fully-copied view of every provenanceKey s
|
||||
// currently holds, safe to read without s's lock — writeProvenanceMetrics
|
||||
// and buildAdminProvenanceViews (admin.go) both read through this rather
|
||||
// than s.data directly.
|
||||
func (s *provenanceStore) snapshot() []provenanceSnapshot {
|
||||
if s == nil {
|
||||
return nil
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
out := make([]provenanceSnapshot, 0, len(s.data))
|
||||
for k, c := range s.data {
|
||||
out = append(out, provenanceSnapshot{key: k, requests: c.requests, tokens: c.tokens})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// recordUsageProvenance accumulates one candidate outcome into
|
||||
// g.provenance — runUnified's (routes_unified.go) only write path into
|
||||
// it, reached only for a candidate whose own response actually completed
|
||||
// (callErr == nil), gated on metricsEnabled exactly like
|
||||
// withLatencyRecorder's own wiring (routes_unified.go): a deployment
|
||||
// with metrics off calls this exactly zero times.
|
||||
func (g *Gateway) recordUsageProvenance(provider, provenance string, tokens int64) {
|
||||
g.provenance.record(provider, provenance, tokens)
|
||||
}
|
||||
|
||||
// writeProvenanceMetrics emits llmgateway_usage_provenance_requests_total
|
||||
// and llmgateway_usage_provenance_tokens_total — this feature's own two
|
||||
// counter families. Absent entirely (no # HELP/TYPE lines emitted) when
|
||||
// g.provenance has recorded nothing yet, mirroring writeLatencyMetrics'
|
||||
// own "no series until the first observation" convention above.
|
||||
//
|
||||
// snaps is sorted (provider, then provenance) purely for a deterministic
|
||||
// scrape body across repeated scrapes of identical state, matching
|
||||
// writeLatencyMetrics' own reasoning.
|
||||
func (g *Gateway) writeProvenanceMetrics(m *metricWriter) {
|
||||
snaps := g.provenance.snapshot()
|
||||
if len(snaps) == 0 {
|
||||
return
|
||||
}
|
||||
sort.Slice(snaps, func(i, j int) bool {
|
||||
a, b := snaps[i].key, snaps[j].key
|
||||
if a.provider != b.provider {
|
||||
return a.provider < b.provider
|
||||
}
|
||||
return a.provenance < b.provenance
|
||||
})
|
||||
|
||||
m.family("llmgateway_usage_provenance_requests_total", "counter",
|
||||
"Total accounted candidate outcomes (a successful upstream response, callErr == nil, that reached usage accounting in runUnified, routes_unified.go), by provider and provenance. provenance is \"reported\" (the provider returned real usage), \"estimated\" (a non-streaming response carried none, so prompt was substituted with ceil(len(body)/4) and completion billed as zero — usage.estimated, limits.go), or \"unbilled\" (a streaming response carried none, so the request is counted but zero tokens are billed). This family only reports how an already-billed number was arrived at; it never changes what gets billed."+aggregationNotePerProcess)
|
||||
m.family("llmgateway_usage_provenance_tokens_total", "counter",
|
||||
"Total tokens billed under each provenance (the SAME result.total() runUnified already passed to accounting, routes_unified.go — never a second, independent count), by provider and provenance. sum(...{provenance=\"reported\"}) / sum(...) across every provenance is the fraction of billed tokens that came from real provider-reported usage."+aggregationNotePerProcess)
|
||||
for _, s := range snaps {
|
||||
labels := []metricLabel{{"provider", s.key.provider}, {"provenance", s.key.provenance}}
|
||||
m.sampleInt("llmgateway_usage_provenance_requests_total", labels, s.requests)
|
||||
m.sampleInt("llmgateway_usage_provenance_tokens_total", labels, s.tokens)
|
||||
}
|
||||
}
|
||||
|
||||
// writeRejectionMetrics emits llmgateway_rate_limit_rejections_total from
|
||||
// limiter.rejections (limits.go) — the one metric family in this file
|
||||
// with no existing counterStore-backed accounting to read: checkAndCount
|
||||
|
||||
+119
@@ -493,6 +493,8 @@ func TestMetrics_OutputParsesAsValidExposition(t *testing.T) {
|
||||
"llmgateway_limit_store_up",
|
||||
"llmgateway_upstream_ttfb_seconds",
|
||||
"llmgateway_upstream_duration_seconds",
|
||||
"llmgateway_usage_provenance_requests_total",
|
||||
"llmgateway_usage_provenance_tokens_total",
|
||||
}
|
||||
for _, name := range wantFamilies {
|
||||
if _, ok := types[name]; !ok {
|
||||
@@ -1334,6 +1336,123 @@ func TestMetrics_LatencyHistogram_StreamingAndNonStreaming_DistinctLabelSets(t *
|
||||
}
|
||||
}
|
||||
|
||||
// --- usage provenance (feat: expose token-accounting provenance) ---
|
||||
|
||||
// TestMetrics_UsageProvenance_EstimatedAndUnbilled_DistinctSeries proves
|
||||
// this feature's central correctness requirement end to end, through the
|
||||
// real /metrics exposition: a non-streaming zero-usage response
|
||||
// (estimated) and a streaming zero-usage response (unbilled) against the
|
||||
// SAME provider render as two DISTINCT series, never merged into one
|
||||
// "not reported" signal. Distinct provider names would make this trivial
|
||||
// (two unrelated series); using the identical provider for both is the
|
||||
// stronger proof — the two provenances would collide into a single
|
||||
// series if the recording code ever conflated them, exactly the
|
||||
// regression this test exists to catch.
|
||||
//
|
||||
// MUTATION VERIFIED: changing routes_unified.go's recording switch case
|
||||
// `case streaming && result.total() == 0:` to `case result.estimated:`
|
||||
// (merging the unbilled classification into the estimated one) made this
|
||||
// test fail: the "unbilled" series vanished from /metrics entirely and
|
||||
// the "estimated" series' requests count doubled to 2, proving this test
|
||||
// actually catches the two provenances collapsing into one — the exact
|
||||
// regression this feature exists to prevent. Reverted before committing.
|
||||
func TestMetrics_UsageProvenance_EstimatedAndUnbilled_DistinctSeries(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
var body map[string]any
|
||||
_ = json.NewDecoder(r.Body).Decode(&body)
|
||||
if streaming, _ := body["stream"].(bool); streaming {
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
fl := w.(http.Flusher)
|
||||
_, _ = fmt.Fprint(w, "data: {\"id\":\"c\",\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n")
|
||||
fl.Flush()
|
||||
_, _ = fmt.Fprint(w, "data: [DONE]\n\n") // no usage chunk: unbilled
|
||||
fl.Flush()
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
// No "usage" field at all: estimated.
|
||||
_, _ = w.Write([]byte(`{"id":"c1","object":"chat.completion","model":"gpt-test","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
cfg := newMetricsTestConfig()
|
||||
cfg.Providers["openai"].BaseURL = srv.URL
|
||||
h, _ := newMetricsGatewayHandle(t, cfg)
|
||||
|
||||
nonStreamBody := map[string]any{"model": "gpt-test", "messages": []any{map[string]any{"role": "user", "content": "hi"}}}
|
||||
streamBody := map[string]any{"model": "gpt-test", "stream": true, "messages": []any{map[string]any{"role": "user", "content": "hi"}}}
|
||||
|
||||
rec1 := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec1, newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", nonStreamBody))
|
||||
if rec1.Code != http.StatusOK {
|
||||
t.Fatalf("non-streaming request status = %d, want 200, body=%s", rec1.Code, rec1.Body.String())
|
||||
}
|
||||
rec2 := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec2, newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", streamBody))
|
||||
if rec2.Code != http.StatusOK {
|
||||
t.Fatalf("streaming request status = %d, want 200, body=%s", rec2.Code, rec2.Body.String())
|
||||
}
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, metricsRequest(http.MethodGet, metricsPathDefault, "sk-admin1"))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("GET /metrics status = %d, want 200, body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
_, samples := parsePrometheusText(t, rec.Body.Bytes())
|
||||
|
||||
var estimatedReq, unbilledReq *promSample
|
||||
for i := range samples {
|
||||
s := &samples[i]
|
||||
if s.name != "llmgateway_usage_provenance_requests_total" || s.labels["provider"] != "openai" {
|
||||
continue
|
||||
}
|
||||
switch s.labels["provenance"] {
|
||||
case provenanceEstimated:
|
||||
estimatedReq = s
|
||||
case provenanceUnbilled:
|
||||
unbilledReq = s
|
||||
}
|
||||
}
|
||||
if estimatedReq == nil {
|
||||
t.Fatal(`no llmgateway_usage_provenance_requests_total{provider="openai",provenance="estimated"} sample — the non-streaming zero-usage request must land under "estimated"`)
|
||||
}
|
||||
if unbilledReq == nil {
|
||||
t.Fatal(`no llmgateway_usage_provenance_requests_total{provider="openai",provenance="unbilled"} sample — the streaming zero-usage request must land under "unbilled", DISTINCT from "estimated"`)
|
||||
}
|
||||
if estimatedReq.value != "1" {
|
||||
t.Errorf(`estimated requests = %s, want 1`, estimatedReq.value)
|
||||
}
|
||||
if unbilledReq.value != "1" {
|
||||
t.Errorf(`unbilled requests = %s, want 1`, unbilledReq.value)
|
||||
}
|
||||
|
||||
var estimatedTok, unbilledTok *promSample
|
||||
for i := range samples {
|
||||
s := &samples[i]
|
||||
if s.name != "llmgateway_usage_provenance_tokens_total" || s.labels["provider"] != "openai" {
|
||||
continue
|
||||
}
|
||||
switch s.labels["provenance"] {
|
||||
case provenanceEstimated:
|
||||
estimatedTok = s
|
||||
case provenanceUnbilled:
|
||||
unbilledTok = s
|
||||
}
|
||||
}
|
||||
if estimatedTok == nil || unbilledTok == nil {
|
||||
t.Fatalf("llmgateway_usage_provenance_tokens_total missing estimated/unbilled series: estimated=%v unbilled=%v", estimatedTok, unbilledTok)
|
||||
}
|
||||
if estimatedTok.value == "0" {
|
||||
t.Error(`estimated tokens = 0, want > 0 — estimated substitutes a nonzero prompt count, unlike unbilled`)
|
||||
}
|
||||
if unbilledTok.value != "0" {
|
||||
t.Errorf(`unbilled tokens = %s, want 0 — unbilled charges nothing at all, unlike estimated`, unbilledTok.value)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRecordLatency_MetricsDisabled_NoObservationCollected proves the
|
||||
// "no collection cost, no series emitted" requirement at the
|
||||
// Gateway/collection level (not merely "the renderer skips an empty
|
||||
|
||||
@@ -555,6 +555,34 @@ func (g *Gateway) runMeteredCall(sw *statusTrackingWriter, r *http.Request, scop
|
||||
g.logf("%s: usage for model %q logged as estimated (%d prompt tokens derived from request body size, not the provider's reported usage)", logPrefix, cand.canonical, result.prompt)
|
||||
}
|
||||
|
||||
// Usage provenance (feat: expose token-accounting provenance),
|
||||
// recorded ONLY for a candidate whose own response actually
|
||||
// completed (callErr == nil): a candidate that failed over never
|
||||
// bills more than a zero-total no-op (this function's own doc
|
||||
// comment above), and counting THAT as "unbilled" would drown the
|
||||
// one signal this feature exists to surface — a provider silently
|
||||
// serving completions free against every budget — in ordinary
|
||||
// failover noise. "unbilled" is re-derived from streaming &&
|
||||
// result.total() == 0 rather than a separate flag carried out of
|
||||
// the block above: result.prompt/result.estimated are only ever
|
||||
// mutated by that block's non-streaming branch, so this condition
|
||||
// is true here if and only if the streaming branch above ran.
|
||||
// Gated on metricsEnabled exactly like withLatencyRecorder's own
|
||||
// wiring above, so a deployment with metrics off pays nothing
|
||||
// extra here. Observation only: nothing recorded here feeds back
|
||||
// into result or the account call above, both of which already
|
||||
// ran — see g.recordUsageProvenance's own doc comment (metrics.go).
|
||||
if callErr == nil && metricsEnabled(g.cfg) {
|
||||
provenance := provenanceReported
|
||||
switch {
|
||||
case result.estimated:
|
||||
provenance = provenanceEstimated
|
||||
case streaming && result.total() == 0:
|
||||
provenance = provenanceUnbilled
|
||||
}
|
||||
g.recordUsageProvenance(providerName, provenance, result.total())
|
||||
}
|
||||
|
||||
// A miss stores the response after everything above has already
|
||||
// run — accounting must never be skipped or delayed waiting on a
|
||||
// cache write. Only a genuine upstream 200 is stored (spec §2's
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
@@ -228,6 +229,250 @@ func TestHandleChat_Streaming_FlushesIncrementally(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// --- usage provenance (feat: expose token-accounting provenance) ---
|
||||
//
|
||||
// Token accounting has three provenances (usage.estimated's own doc
|
||||
// comment, limits.go): reported (real provider usage), estimated (a
|
||||
// non-streaming response with no usage, prompt substituted from body
|
||||
// size), and unbilled (a streaming response with no usage, request
|
||||
// counted but zero tokens billed). These three tests drive each real
|
||||
// HTTP shape through Gateway.ServeHTTP and assert both the provenance
|
||||
// classification AND that billing itself is byte-identical to the
|
||||
// pre-feature contract — this feature must never change a billed number.
|
||||
|
||||
// TestUsageProvenance_ReportedUsage_ClassifiedReported proves the normal,
|
||||
// accurate case (a provider that returns real usage) is classified
|
||||
// "reported", not merely "not estimated" by omission.
|
||||
//
|
||||
// MUTATION VERIFIED: changing routes_unified.go's provenance switch
|
||||
// default from `provenance := provenanceReported` to `provenance :=
|
||||
// provenanceEstimated` made this test fail — the seen provenance became
|
||||
// "estimated" instead of "reported". Reverted before committing.
|
||||
func TestUsageProvenance_ReportedUsage_ClassifiedReported(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write([]byte(`{"id":"c1","object":"chat.completion","model":"gpt-test","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],"usage":{"prompt_tokens":10,"completion_tokens":5}}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
cfg := CreateConfig()
|
||||
cfg.Metrics = &MetricsConfig{Enabled: true}
|
||||
cfg.Providers = map[string]*ProviderConfig{
|
||||
"openai": {Type: "openai", BaseURL: srv.URL, APIKey: "sk-up", Models: []string{"gpt-test"}},
|
||||
}
|
||||
cfg.Groups = map[string]*GroupConfig{"default": {}}
|
||||
cfg.Users = &UsersConfig{Inline: []*UserConfig{{Name: "alice", Group: "default", APIKey: "sk-alice"}}}
|
||||
next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})
|
||||
h, err := New(context.Background(), next, cfg, "llmgw")
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
gw, ok := h.(*Gateway)
|
||||
if !ok {
|
||||
t.Fatal("handler is not *Gateway")
|
||||
}
|
||||
|
||||
body := map[string]any{"model": "gpt-test", "messages": []any{map[string]any{"role": "user", "content": "hi"}}}
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200, body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
|
||||
snaps := gw.provenance.snapshot()
|
||||
if len(snaps) != 1 {
|
||||
t.Fatalf("g.provenance has %d entries, want 1: %+v", len(snaps), snaps)
|
||||
}
|
||||
s := snaps[0]
|
||||
if s.key.provider != "openai" || s.key.provenance != provenanceReported {
|
||||
t.Errorf("provenance key = %+v, want provider=openai provenance=reported", s.key)
|
||||
}
|
||||
if s.requests != 1 {
|
||||
t.Errorf("requests = %d, want 1", s.requests)
|
||||
}
|
||||
if s.tokens != 15 {
|
||||
t.Errorf("tokens = %d, want 15 (10 prompt + 5 completion, the real reported usage)", s.tokens)
|
||||
}
|
||||
|
||||
// Billing itself must be exactly what TestHandleChat_HappyPath_
|
||||
// NonStreaming_AccountsUsage already proves for this identical
|
||||
// response shape — this feature must never change a billed number.
|
||||
tokIn, _ := gw.limiter.getCounter("user", "alice", metricTokIn, windowDay, time.Now())
|
||||
tokOut, _ := gw.limiter.getCounter("user", "alice", metricTokOut, windowDay, time.Now())
|
||||
if tokIn != 10 || tokOut != 5 {
|
||||
t.Errorf("billed tokin/tokout = %d/%d, want 10/5 — provenance recording must never change what gets billed", tokIn, tokOut)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUsageProvenance_NonStreamingZeroUsage_ClassifiedEstimated proves a
|
||||
// non-streaming response that carries no usage at all is classified
|
||||
// "estimated", with the SAME prompt-estimate/zero-completion billing
|
||||
// routes_unified.go already applied before this feature existed —
|
||||
// ceil(request body size / 4) prompt, zero completion.
|
||||
//
|
||||
// MUTATION VERIFIED: changing the recording switch's `case
|
||||
// result.estimated:` guard (routes_unified.go) to `case false:` made
|
||||
// this test fail — the seen provenance stayed "reported" instead of
|
||||
// becoming "estimated". Reverted before committing.
|
||||
func TestUsageProvenance_NonStreamingZeroUsage_ClassifiedEstimated(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
// No "usage" field at all — decodes to chatUsagePayload's zero
|
||||
// value (provider_openai.go's forwardJSON), the identical shape a
|
||||
// provider that omits usage from a non-streaming response sends.
|
||||
_, _ = w.Write([]byte(`{"id":"c1","object":"chat.completion","model":"gpt-test","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
cfg := CreateConfig()
|
||||
cfg.Metrics = &MetricsConfig{Enabled: true}
|
||||
cfg.Providers = map[string]*ProviderConfig{
|
||||
"openai": {Type: "openai", BaseURL: srv.URL, APIKey: "sk-up", Models: []string{"gpt-test"}},
|
||||
}
|
||||
cfg.Groups = map[string]*GroupConfig{"default": {}}
|
||||
cfg.Users = &UsersConfig{Inline: []*UserConfig{{Name: "alice", Group: "default", APIKey: "sk-alice"}}}
|
||||
next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})
|
||||
h, err := New(context.Background(), next, cfg, "llmgw")
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
gw, ok := h.(*Gateway)
|
||||
if !ok {
|
||||
t.Fatal("handler is not *Gateway")
|
||||
}
|
||||
|
||||
reqBody := map[string]any{"model": "gpt-test", "messages": []any{map[string]any{"role": "user", "content": "hi"}}}
|
||||
// The exact bytes newUnifiedRequest below marshals reqBody into —
|
||||
// routes_unified.go's own estimate is ceil(len(body) / 4) over these
|
||||
// SAME bytes, so this reproduces the source's formula against a
|
||||
// black-box-observable input (the request the test itself sends),
|
||||
// not a re-implementation of internal logic.
|
||||
marshaled, err := json.Marshal(reqBody)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal request body: %v", err)
|
||||
}
|
||||
wantPrompt := int64(math.Ceil(float64(len(marshaled)) / 4))
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
h.ServeHTTP(rec, newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", reqBody))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200, body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
|
||||
snaps := gw.provenance.snapshot()
|
||||
if len(snaps) != 1 {
|
||||
t.Fatalf("g.provenance has %d entries, want 1: %+v", len(snaps), snaps)
|
||||
}
|
||||
s := snaps[0]
|
||||
if s.key.provider != "openai" || s.key.provenance != provenanceEstimated {
|
||||
t.Errorf("provenance key = %+v, want provider=openai provenance=estimated", s.key)
|
||||
}
|
||||
if s.tokens != wantPrompt {
|
||||
t.Errorf("provenance tokens = %d, want %d (ceil(request body size / 4), the estimate)", s.tokens, wantPrompt)
|
||||
}
|
||||
|
||||
// Billing unchanged from the pre-feature contract: prompt substituted
|
||||
// with the estimate, completion billed as zero.
|
||||
tokIn, _ := gw.limiter.getCounter("user", "alice", metricTokIn, windowDay, time.Now())
|
||||
tokOut, _ := gw.limiter.getCounter("user", "alice", metricTokOut, windowDay, time.Now())
|
||||
if tokIn != wantPrompt {
|
||||
t.Errorf("billed tokin = %d, want %d (ceil(request body size / 4))", tokIn, wantPrompt)
|
||||
}
|
||||
if tokOut != 0 {
|
||||
t.Errorf("billed tokout = %d, want 0 — completion is billed as zero for an estimated response", tokOut)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUsageProvenance_StreamingZeroUsage_ClassifiedUnbilled_DistinctFromEstimated
|
||||
// proves a streaming response that carries no usage at all is classified
|
||||
// "unbilled" — DISTINCT from "estimated" (this feature's own
|
||||
// correctness requirement: collapsing the two into one "not reported"
|
||||
// signal would hide that a provider serving completions entirely free
|
||||
// is a materially different failure than an approximated bill) — and
|
||||
// that billing stays exactly what routes_unified.go already applied
|
||||
// before this feature existed: the request counted, zero tokens billed.
|
||||
//
|
||||
// MUTATION VERIFIED: changing the recording switch's `case streaming &&
|
||||
// result.total() == 0:` guard (routes_unified.go) to `case
|
||||
// result.estimated:` (merging unbilled into the estimated branch) made
|
||||
// this test fail — the seen provenance became "estimated" instead of
|
||||
// "unbilled", and the distinctness assertion below failed too. Reverted
|
||||
// before committing.
|
||||
func TestUsageProvenance_StreamingZeroUsage_ClassifiedUnbilled_DistinctFromEstimated(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
fl, ok := w.(http.Flusher)
|
||||
if !ok {
|
||||
t.Fatal("fake upstream ResponseWriter does not support Flush")
|
||||
}
|
||||
_, _ = w.Write([]byte("data: {\"id\":\"c1\",\"object\":\"chat.completion.chunk\",\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n"))
|
||||
fl.Flush()
|
||||
// No usage-only chunk — the provider never honored
|
||||
// stream_options.include_usage, the exact shape this feature
|
||||
// exists to surface.
|
||||
_, _ = w.Write([]byte("data: [DONE]\n\n"))
|
||||
fl.Flush()
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
cfg := CreateConfig()
|
||||
cfg.Metrics = &MetricsConfig{Enabled: true}
|
||||
cfg.Providers = map[string]*ProviderConfig{
|
||||
"openai": {Type: "openai", BaseURL: srv.URL, APIKey: "sk-up", Models: []string{"gpt-test"}},
|
||||
}
|
||||
cfg.Groups = map[string]*GroupConfig{"default": {}}
|
||||
cfg.Users = &UsersConfig{Inline: []*UserConfig{{Name: "alice", Group: "default", APIKey: "sk-alice"}}}
|
||||
next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})
|
||||
h, err := New(context.Background(), next, cfg, "llmgw")
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
gw, ok := h.(*Gateway)
|
||||
if !ok {
|
||||
t.Fatal("handler is not *Gateway")
|
||||
}
|
||||
|
||||
body := map[string]any{"model": "gpt-test", "stream": true, "messages": []any{map[string]any{"role": "user", "content": "hi"}}}
|
||||
rw := newRecordingWriter()
|
||||
h.ServeHTTP(rw, newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body))
|
||||
if rw.status != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200", rw.status)
|
||||
}
|
||||
|
||||
snaps := gw.provenance.snapshot()
|
||||
if len(snaps) != 1 {
|
||||
t.Fatalf("g.provenance has %d entries, want 1: %+v", len(snaps), snaps)
|
||||
}
|
||||
s := snaps[0]
|
||||
if s.key.provider != "openai" || s.key.provenance != provenanceUnbilled {
|
||||
t.Errorf("provenance key = %+v, want provider=openai provenance=unbilled", s.key)
|
||||
}
|
||||
if s.key.provenance == provenanceEstimated {
|
||||
t.Errorf("provenance = estimated, want unbilled — a streaming zero-usage response must never be classified the same as a non-streaming one")
|
||||
}
|
||||
if s.tokens != 0 {
|
||||
t.Errorf("provenance tokens = %d, want 0 — unbilled charges nothing at all", s.tokens)
|
||||
}
|
||||
|
||||
// Billing unchanged from the pre-feature contract: the request is
|
||||
// counted (req/min bumped) but zero tokens are billed.
|
||||
tokIn, _ := gw.limiter.getCounter("user", "alice", metricTokIn, windowDay, time.Now())
|
||||
tokOut, _ := gw.limiter.getCounter("user", "alice", metricTokOut, windowDay, time.Now())
|
||||
if tokIn != 0 || tokOut != 0 {
|
||||
t.Errorf("billed tokin/tokout = %d/%d, want 0/0 — a streaming zero-usage response bills nothing", tokIn, tokOut)
|
||||
}
|
||||
reqCount, ok2 := gw.limiter.getCounter("user", "alice", metricReq, windowMin, time.Now())
|
||||
if !ok2 || reqCount != 1 {
|
||||
t.Errorf("request/min counter = %d (ok=%v), want 1 — the request must still be COUNTED even though zero tokens are billed", reqCount, ok2)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHandleChat_NoAPIKey_Returns401 asserts the unified routes require
|
||||
// authentication before any resolve/limit work happens.
|
||||
func TestHandleChat_NoAPIKey_Returns401(t *testing.T) {
|
||||
|
||||
+198
-8
@@ -164,6 +164,26 @@ const (
|
||||
hungBodyUpstreamSleep = 600 * time.Millisecond
|
||||
)
|
||||
|
||||
// provenanceEstProviderName/provenanceEstProviderModel back
|
||||
// exerciseUsageProvenance (feat: expose token-accounting provenance): a
|
||||
// provider whose upstream answers 200 with NO "usage" field at all —
|
||||
// decodes to chatUsagePayload's zero value (provider_openai.go's
|
||||
// forwardJSON), the exact shape runUnified's own estimation fallback
|
||||
// (routes_unified.go) classifies as "estimated" — proving the
|
||||
// substitution math (ceil(len(body)/4)) and the new
|
||||
// llmgateway_usage_provenance_requests_total/-_tokens_total families
|
||||
// (metrics.go) all run correctly under the REAL interpreter, not merely
|
||||
// compiled. "reported" is already exercised, under the interpreter, by
|
||||
// exerciseAttemptAccounting's own real, usage-bearing "openai" traffic
|
||||
// above — no separate fixture needed for that provenance here.
|
||||
const (
|
||||
provenanceEstProviderName = "provenance-est"
|
||||
// provenanceEstProviderModel: same reasoning as timeoutProviderModel's
|
||||
// own doc comment above — must not collide with testDataWantModel or
|
||||
// any other probe's model id.
|
||||
provenanceEstProviderModel = "gpt-provenance-est-test"
|
||||
)
|
||||
|
||||
// excludedTopLevelDirs lists repo-root directories the GOPATH copy must
|
||||
// never include: build tooling, integration fixtures, planning docs and
|
||||
// VCS metadata have nothing to do with the plugin package Yaegi imports.
|
||||
@@ -348,6 +368,18 @@ func run() error {
|
||||
}))
|
||||
defer hungBodyUpstream.Close()
|
||||
|
||||
// provenanceEstUpstream backs exerciseUsageProvenance (feat: expose
|
||||
// token-accounting provenance): answers 200 with NO "usage" field at
|
||||
// all — see provenanceEstProviderName's own const block doc comment
|
||||
// for why this is the exact shape runUnified's estimation fallback
|
||||
// classifies as "estimated".
|
||||
provenanceEstUpstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write([]byte(`{"id":"c-prov","object":"chat.completion","model":"` + provenanceEstProviderModel + `","choices":[{"index":0,"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}]}`))
|
||||
}))
|
||||
defer provenanceEstUpstream.Close()
|
||||
|
||||
// mcpProbeUpstream answers federation's outbound tools/call with a
|
||||
// body deliberately larger than mcpBackendCallResponseMaxBytes, so
|
||||
// exerciseHandler's POST /mcp probe below drives doBackendJSONRPC's
|
||||
@@ -492,6 +524,9 @@ func run() error {
|
||||
// comment above for why this provider exists alongside
|
||||
// timeoutProviderName.
|
||||
`"` + hungBodyProviderName + `":{"type":"openai","baseUrl":"` + hungBodyUpstream.URL + `","apiKey":"sk-up","requestTimeout":"` + hungBodyProviderRequestTimeout + `","models":["` + hungBodyProviderModel + `"]},` +
|
||||
// provenanceEstProviderName (feat: expose token-accounting
|
||||
// provenance) — see its own const block doc comment above.
|
||||
`"` + provenanceEstProviderName + `":{"type":"openai","baseUrl":"` + provenanceEstUpstream.URL + `","apiKey":"sk-up","models":["` + provenanceEstProviderModel + `"]},` +
|
||||
// "anthropic" backs the /v1/messages passthrough probes
|
||||
// (exerciseMessagesRoute, below): a real anthropic-type provider,
|
||||
// interpreted end to end through
|
||||
@@ -591,14 +626,21 @@ func run() error {
|
||||
return err
|
||||
}
|
||||
// exerciseLatencyMetrics (feat: instrument upstream latency) runs
|
||||
// LAST, for the same reason exerciseMetricsRoute itself used to be
|
||||
// last: it reads the exact same /metrics endpoint and must not race
|
||||
// or perturb anything exerciseBreaker/exerciseMetricsRoute still care
|
||||
// about. It relies on exerciseHandler's own exerciseAttemptAccounting
|
||||
// sub-probe (already run, above) having driven a real, non-streaming
|
||||
// POST /v1/chat/completions against the "openai" provider — see its
|
||||
// own doc comment.
|
||||
return exerciseLatencyMetrics(handler)
|
||||
// after exerciseMetricsRoute, for the same reason exerciseMetricsRoute
|
||||
// itself used to be last: it reads the exact same /metrics endpoint
|
||||
// and must not race or perturb anything exerciseBreaker/
|
||||
// exerciseMetricsRoute still care about. It relies on exerciseHandler's
|
||||
// own exerciseAttemptAccounting sub-probe (already run, above) having
|
||||
// driven a real, non-streaming POST /v1/chat/completions against the
|
||||
// "openai" provider — see its own doc comment.
|
||||
if err := exerciseLatencyMetrics(handler); err != nil {
|
||||
return err
|
||||
}
|
||||
// exerciseUsageProvenance (feat: expose token-accounting provenance)
|
||||
// runs LAST, for the identical reason: it reads /metrics and GET
|
||||
// /admin/api/overview one more time and must not perturb any counter
|
||||
// an earlier probe already asserted on.
|
||||
return exerciseUsageProvenance(handler)
|
||||
}
|
||||
|
||||
// builtinLookupModelID is a real, stable entry in the generated
|
||||
@@ -1035,6 +1077,154 @@ func exerciseLatencyMetrics(handler http.Handler) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// exerciseUsageProvenance proves the usage-provenance metric surface
|
||||
// (feat: expose token-accounting provenance, metrics.go's
|
||||
// provenanceStore/writeProvenanceMetrics, admin.go's
|
||||
// buildAdminProvenanceViews) renders correctly under the REAL
|
||||
// interpreter — not merely that a compiled `go test` accepts the same
|
||||
// bytes, this package's own doc comment's exact concern.
|
||||
//
|
||||
// Drives one real, non-streaming POST /v1/chat/completions against
|
||||
// provenanceEstProviderName, whose upstream (provenanceEstUpstream, run())
|
||||
// answers 200 with no "usage" field at all — the exact shape
|
||||
// runUnified's estimation fallback (routes_unified.go) classifies as
|
||||
// "estimated". "reported" is not driven fresh here: exerciseHandler's own
|
||||
// exerciseAttemptAccounting sub-probe (already run, above) already drove
|
||||
// a real, usage-bearing POST /v1/chat/completions against "openai",
|
||||
// which is reported provenance by construction — reusing that traffic
|
||||
// instead of a third fixture, mirroring exerciseLatencyMetrics' own
|
||||
// "no new traffic" reasoning for its own duration-histogram assertion.
|
||||
func exerciseUsageProvenance(handler http.Handler) error {
|
||||
chatReq := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(
|
||||
`{"model":"`+provenanceEstProviderName+`/`+provenanceEstProviderModel+`","messages":[{"role":"user","content":"hi"}]}`,
|
||||
))
|
||||
chatReq.Header.Set("Authorization", "Bearer "+testDataUserAPIKey)
|
||||
chatReq.Header.Set("Content-Type", "application/json")
|
||||
chatRec := httptest.NewRecorder()
|
||||
handler.ServeHTTP(chatRec, chatReq)
|
||||
if chatRec.Code != http.StatusOK {
|
||||
return fmt.Errorf("POST /v1/chat/completions (usage-provenance harness): status = %d, want 200, body=%s", chatRec.Code, chatRec.Body.String())
|
||||
}
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
|
||||
req.RemoteAddr = metricsProbeOutsideAddr
|
||||
req.Header.Set("Authorization", "Bearer "+attemptAccountingAdminAPIKey)
|
||||
rec := httptest.NewRecorder()
|
||||
handler.ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusOK {
|
||||
return fmt.Errorf("GET /metrics (usage-provenance probe): status = %d, want 200, body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
|
||||
parser := expfmt.NewTextParser(model.LegacyValidation)
|
||||
families, err := parser.TextToMetricFamilies(strings.NewReader(rec.Body.String()))
|
||||
if err != nil {
|
||||
return fmt.Errorf("GET /metrics: a real Prometheus TextParser rejected the interpreted rendering: %w (body=%s)", err, rec.Body.String())
|
||||
}
|
||||
|
||||
requestsFamily, ok := families["llmgateway_usage_provenance_requests_total"]
|
||||
if !ok {
|
||||
return fmt.Errorf("GET /metrics: no llmgateway_usage_provenance_requests_total family in the parsed output (body=%s)", rec.Body.String())
|
||||
}
|
||||
if requestsFamily.GetType() != dto.MetricType_COUNTER {
|
||||
return fmt.Errorf("llmgateway_usage_provenance_requests_total type = %s, want COUNTER", requestsFamily.GetType())
|
||||
}
|
||||
tokensFamily, ok := families["llmgateway_usage_provenance_tokens_total"]
|
||||
if !ok {
|
||||
return fmt.Errorf("GET /metrics: no llmgateway_usage_provenance_tokens_total family in the parsed output (body=%s)", rec.Body.String())
|
||||
}
|
||||
if tokensFamily.GetType() != dto.MetricType_COUNTER {
|
||||
return fmt.Errorf("llmgateway_usage_provenance_tokens_total type = %s, want COUNTER", tokensFamily.GetType())
|
||||
}
|
||||
|
||||
findSample := func(fam *dto.MetricFamily, provider, provenance string) *dto.Metric {
|
||||
for _, m := range fam.GetMetric() {
|
||||
var p, k string
|
||||
for _, lp := range m.GetLabel() {
|
||||
switch lp.GetName() {
|
||||
case "provider":
|
||||
p = lp.GetValue()
|
||||
case "provenance":
|
||||
k = lp.GetValue()
|
||||
}
|
||||
}
|
||||
if p == provider && k == provenance {
|
||||
return m
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
estReq := findSample(requestsFamily, provenanceEstProviderName, "estimated")
|
||||
if estReq == nil {
|
||||
return fmt.Errorf(`GET /metrics: no llmgateway_usage_provenance_requests_total{provider=%q,provenance="estimated"} sample found (body=%s)`, provenanceEstProviderName, rec.Body.String())
|
||||
}
|
||||
if estReq.GetCounter().GetValue() != 1 {
|
||||
return fmt.Errorf(`llmgateway_usage_provenance_requests_total{provider=%q,provenance="estimated"} = %v, want 1`, provenanceEstProviderName, estReq.GetCounter().GetValue())
|
||||
}
|
||||
estTok := findSample(tokensFamily, provenanceEstProviderName, "estimated")
|
||||
if estTok == nil {
|
||||
return fmt.Errorf(`GET /metrics: no llmgateway_usage_provenance_tokens_total{provider=%q,provenance="estimated"} sample found (body=%s)`, provenanceEstProviderName, rec.Body.String())
|
||||
}
|
||||
if estTok.GetCounter().GetValue() <= 0 {
|
||||
return fmt.Errorf(`llmgateway_usage_provenance_tokens_total{provider=%q,provenance="estimated"} = %v, want > 0 (estimated substitutes a nonzero prompt count)`, provenanceEstProviderName, estTok.GetCounter().GetValue())
|
||||
}
|
||||
|
||||
reportedReq := findSample(requestsFamily, "openai", "reported")
|
||||
if reportedReq == nil {
|
||||
return fmt.Errorf(`GET /metrics: no llmgateway_usage_provenance_requests_total{provider="openai",provenance="reported"} sample found (body=%s) — exerciseAttemptAccounting's own earlier real, usage-bearing traffic must have been classified "reported"`, rec.Body.String())
|
||||
}
|
||||
if reportedReq.GetCounter().GetValue() < 1 {
|
||||
return fmt.Errorf(`llmgateway_usage_provenance_requests_total{provider="openai",provenance="reported"} = %v, want >= 1`, reportedReq.GetCounter().GetValue())
|
||||
}
|
||||
|
||||
// GET /admin/api/overview: proves admin.go's buildAdminProvenanceViews
|
||||
// wiring also runs correctly interpreted, decoded generically (this
|
||||
// harness module is compiled, not interpreted; adminOverviewResponse/
|
||||
// adminProviderView are unexported — see readProviderAttemptCounters'
|
||||
// own doc comment for why a generic decode is simpler here than
|
||||
// exporting test-only types across that boundary).
|
||||
adminReq := httptest.NewRequest(http.MethodGet, "/admin/api/overview", nil)
|
||||
adminReq.Header.Set("Authorization", "Bearer "+attemptAccountingAdminAPIKey)
|
||||
adminRec := httptest.NewRecorder()
|
||||
handler.ServeHTTP(adminRec, adminReq)
|
||||
if adminRec.Code != http.StatusOK {
|
||||
return fmt.Errorf("GET /admin/api/overview (usage-provenance probe): status = %d, want 200, body=%s", adminRec.Code, adminRec.Body.String())
|
||||
}
|
||||
var overview map[string]any
|
||||
if err := json.Unmarshal(adminRec.Body.Bytes(), &overview); err != nil {
|
||||
return fmt.Errorf("decode GET /admin/api/overview body: %w", err)
|
||||
}
|
||||
providers, _ := overview["providers"].([]any)
|
||||
var provenanceEstProvider map[string]any
|
||||
for _, p := range providers {
|
||||
pm, ok := p.(map[string]any)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if pm["name"] == provenanceEstProviderName {
|
||||
provenanceEstProvider = pm
|
||||
break
|
||||
}
|
||||
}
|
||||
if provenanceEstProvider == nil {
|
||||
return fmt.Errorf("GET /admin/api/overview: no provider entry named %q (body=%s)", provenanceEstProviderName, adminRec.Body.String())
|
||||
}
|
||||
provenanceField, _ := provenanceEstProvider["provenance"].(map[string]any)
|
||||
estimatedEntry, _ := provenanceField["estimated"].(map[string]any)
|
||||
if estimatedEntry == nil {
|
||||
return fmt.Errorf(`GET /admin/api/overview: provider %q has no provenance.estimated entry: %+v`, provenanceEstProviderName, provenanceField)
|
||||
}
|
||||
if reqs, _ := estimatedEntry["requests"].(float64); reqs != 1 {
|
||||
return fmt.Errorf("GET /admin/api/overview: provider %q provenance.estimated.requests = %v, want 1", provenanceEstProviderName, estimatedEntry["requests"])
|
||||
}
|
||||
if _, hasReported := provenanceField["reported"]; hasReported {
|
||||
return fmt.Errorf(`GET /admin/api/overview: provider %q provenance has a "reported" key, want it excluded from this compact admin summary: %+v`, provenanceEstProviderName, provenanceField)
|
||||
}
|
||||
|
||||
fmt.Println("yaegi-check: usage-provenance metrics (llmgateway_usage_provenance_requests_total/-_tokens_total) and admin overview parsed by a real Prometheus TextParser/JSON decoder; estimated and reported provenances both present and distinct")
|
||||
return nil
|
||||
}
|
||||
|
||||
// exerciseHandler runs two real requests against the interpreted
|
||||
// handler, drawn from the plugin's own .traefik.yml testData: an
|
||||
// authenticated GET /v1/models, which must return the configured
|
||||
|
||||
@@ -25,7 +25,7 @@ import { formatAgo, formatContextWindow, formatLatencyMs, formatUntil, refreshDe
|
||||
import { isModelDegraded } from '@/lib/provider-rate'
|
||||
import { type ExpandState, clearExpandOverrides, computeExpandedItems, toggleItemExpand } from '@/lib/search-expand'
|
||||
import { useDashboardStore } from '@/stores/dashboard'
|
||||
import type { AdminAliasView, AdminLatencyView, AdminModelMetaView, AdminModelRateView, AdminProviderView } from '@/types/api'
|
||||
import type { AdminAliasView, AdminLatencyView, AdminModelMetaView, AdminModelRateView, AdminProvenanceView, AdminProviderView } from '@/types/api'
|
||||
|
||||
// This component backs the "Providers" tab (App.vue) — a UI-label rename
|
||||
// only. The data it renders still comes from GET /admin/api/overview
|
||||
@@ -251,6 +251,40 @@ function nonStreamingLatencyTitle(v: AdminLatencyView): string {
|
||||
return `Average total response time, including generation — not a load-sensitive signal like streaming ttfb, since this provider buffers the whole completion before sending anything. ${latencyObservationCount(v)}. ${PER_REPLICA_CAVEAT}`
|
||||
}
|
||||
|
||||
// --- usage provenance (feat: expose token-accounting provenance) ---
|
||||
//
|
||||
// Token accounting has three provenances (see AdminProvenanceView's own
|
||||
// doc comment): reported (the healthy default — no badge, mirroring the
|
||||
// health/latency badges' own "nothing to show reads as no badge"
|
||||
// convention), estimated (a non-streaming response carried no usage, so
|
||||
// prompt was substituted from request body size), and unbilled (a
|
||||
// streaming response carried no usage, so the request was counted but
|
||||
// zero tokens were billed — a provider potentially serving completions
|
||||
// entirely free against every budget). estimated/unbilled are two
|
||||
// DISTINCT badges, never merged into one "not reported" indicator —
|
||||
// collapsing them would hide exactly the fact this feature exists to
|
||||
// surface (task brief's own correctness rule).
|
||||
|
||||
/** estimatedProvenance/unbilledProvenance pull p.provenance's two known kind keys (admin.go's buildAdminProvenanceViews emits only these two strings, never "reported") — undefined when that kind has no observations yet. */
|
||||
function estimatedProvenance(p: AdminProviderView): AdminProvenanceView | undefined {
|
||||
return p.provenance?.estimated
|
||||
}
|
||||
function unbilledProvenance(p: AdminProviderView): AdminProvenanceView | undefined {
|
||||
return p.provenance?.unbilled
|
||||
}
|
||||
|
||||
/** estimatedProvenanceTitle is the estimated-accounting badge's hover/title detail. */
|
||||
function estimatedProvenanceTitle(v: AdminProvenanceView): string {
|
||||
const plural = v.requests === 1 ? '' : 's'
|
||||
return `${v.requests} non-streaming response${plural} carried no usage from the provider — prompt tokens were estimated from request body size instead (${v.tokens} tokens billed from the estimate), completion billed as zero. ${PER_REPLICA_CAVEAT}`
|
||||
}
|
||||
|
||||
/** unbilledProvenanceTitle is the unbilled-accounting badge's hover/title detail — explicit about the budget-exposure implication (task brief's own framing). */
|
||||
function unbilledProvenanceTitle(v: AdminProvenanceView): string {
|
||||
const plural = v.requests === 1 ? '' : 's'
|
||||
return `${v.requests} streaming response${plural} carried no usage from the provider — the request was counted but zero tokens were billed. If this keeps happening, this provider may be serving completions entirely free against every budget. ${PER_REPLICA_CAVEAT}`
|
||||
}
|
||||
|
||||
// --- model aliases (sortable DataTable, operator directive) ---
|
||||
//
|
||||
// The alias id gets the ModelChip copy treatment (an alias name IS the
|
||||
@@ -419,6 +453,33 @@ const aliasEmptyMessage = computed(() =>
|
||||
>
|
||||
non-streaming avg {{ formatLatencyMs(nonStreamingLatency(p)!.avgDurationMs as number) }}
|
||||
</Badge>
|
||||
<!--
|
||||
Usage provenance (feat: expose token-accounting
|
||||
provenance): no badge at all for a kind with no
|
||||
observations yet, mirroring the health/latency badges'
|
||||
own convention above. estimated and unbilled are two
|
||||
distinct badges — never merged (correctness rule, task
|
||||
brief) — since one substitutes an approximation and the
|
||||
other bills nothing at all.
|
||||
-->
|
||||
<Badge
|
||||
v-if="estimatedProvenance(p)"
|
||||
as="span"
|
||||
variant="secondary"
|
||||
class="font-normal tabular-nums"
|
||||
:title="estimatedProvenanceTitle(estimatedProvenance(p) as AdminProvenanceView)"
|
||||
>
|
||||
estimated {{ estimatedProvenance(p)!.requests }}
|
||||
</Badge>
|
||||
<Badge
|
||||
v-if="unbilledProvenance(p)"
|
||||
as="span"
|
||||
variant="destructive"
|
||||
class="font-normal tabular-nums"
|
||||
:title="unbilledProvenanceTitle(unbilledProvenance(p) as AdminProvenanceView)"
|
||||
>
|
||||
unbilled {{ unbilledProvenance(p)!.requests }}
|
||||
</Badge>
|
||||
<FontAwesomeIcon
|
||||
v-if="p.lastErr"
|
||||
:icon="faTriangleExclamation"
|
||||
|
||||
@@ -58,6 +58,26 @@ export interface AdminLatencyView {
|
||||
count?: number
|
||||
}
|
||||
|
||||
/**
|
||||
* One provider's one non-reported provenance kind's compact usage-
|
||||
* accounting summary (admin.go: adminProvenanceView — feat: expose
|
||||
* token-accounting provenance). "estimated" means a non-streaming
|
||||
* response carried no usage from the provider, so prompt was substituted
|
||||
* with an estimate (ceil(request body length / 4)) and completion billed
|
||||
* as zero. "unbilled" means a streaming response carried no usage, so
|
||||
* the request was counted but zero tokens were billed — a provider
|
||||
* silently ignoring stream_options.include_usage serves that completion
|
||||
* entirely free against every budget. Billing itself is unaffected by
|
||||
* this data; it only reports how an already-billed number was arrived
|
||||
* at. requests/tokens are never fabricated zeros: this view only ever
|
||||
* exists (see AdminProviderView.provenance) for a kind that was actually
|
||||
* observed at least once.
|
||||
*/
|
||||
export interface AdminProvenanceView {
|
||||
requests: number
|
||||
tokens: number
|
||||
}
|
||||
|
||||
export interface AdminProviderView {
|
||||
name: string
|
||||
type: string
|
||||
@@ -142,6 +162,22 @@ export interface AdminProviderView {
|
||||
* together or present either one as a bare, unlabeled "latency".
|
||||
*/
|
||||
latency?: Partial<Record<'streaming' | 'non-streaming', AdminLatencyView>>
|
||||
/**
|
||||
* provenance summarizes non-reported usage accounting for this
|
||||
* provider (admin.go: adminProviderView.Provenance — feat: expose
|
||||
* token-accounting provenance), keyed by provenance kind: 'estimated'
|
||||
* or 'unbilled' only — 'reported' (the default, healthy case) is never
|
||||
* a key here, see AdminProvenanceView's own doc comment for why.
|
||||
* Undefined entirely for a provider with no estimated/unbilled outcome
|
||||
* yet, or a deployment with metrics collection gated off — mirroring
|
||||
* latency's identical "undefined, never an empty object" convention
|
||||
* above.
|
||||
*
|
||||
* PER-REPLICA, IN-PROCESS: sourced from the same in-process accumulator
|
||||
* as latency above — see that field's own caveat, which applies here
|
||||
* unchanged.
|
||||
*/
|
||||
provenance?: Partial<Record<'estimated' | 'unbilled', AdminProvenanceView>>
|
||||
}
|
||||
|
||||
export interface AdminRedisView {
|
||||
|
||||
Reference in New Issue
Block a user