From ba11b9ae4312152aecbb8c843f4f51f5bab694f5 Mon Sep 17 00:00:00 2001 From: roroghost17 Date: Wed, 27 May 2026 19:52:12 +0530 Subject: [PATCH] feat: adds provider cache and semantic cache attributes in metrics export --- core/schemas/mux.go | 22 ++++----- flake.lock | 6 +-- plugins/otel/main.go | 85 +++++++++++++++++++++++++++++++++-- plugins/otel/metrics.go | 55 +++++++++++++++++++++++ plugins/telemetry/main.go | 94 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 246 insertions(+), 16 deletions(-) diff --git a/core/schemas/mux.go b/core/schemas/mux.go index 5168e30013b..b6ac0896577 100644 --- a/core/schemas/mux.go +++ b/core/schemas/mux.go @@ -986,11 +986,12 @@ func (cu *BifrostLLMUsage) ToResponsesResponseUsage() *ResponsesResponseUsage { if cu.PromptTokensDetails != nil { usage.InputTokensDetails = &ResponsesResponseInputTokens{ - TextTokens: cu.PromptTokensDetails.TextTokens, - AudioTokens: cu.PromptTokensDetails.AudioTokens, - ImageTokens: cu.PromptTokensDetails.ImageTokens, - CachedReadTokens: cu.PromptTokensDetails.CachedReadTokens, - CachedWriteTokens: cu.PromptTokensDetails.CachedWriteTokens, + TextTokens: cu.PromptTokensDetails.TextTokens, + AudioTokens: cu.PromptTokensDetails.AudioTokens, + ImageTokens: cu.PromptTokensDetails.ImageTokens, + CachedReadTokens: cu.PromptTokensDetails.CachedReadTokens, + CachedWriteTokens: cu.PromptTokensDetails.CachedWriteTokens, + CachedWriteTokenDetails: cu.PromptTokensDetails.CachedWriteTokenDetails, } } if cu.CompletionTokensDetails != nil { @@ -1022,11 +1023,12 @@ func (ru *ResponsesResponseUsage) ToBifrostLLMUsage() *BifrostLLMUsage { if ru.InputTokensDetails != nil { usage.PromptTokensDetails = &ChatPromptTokensDetails{ - TextTokens: ru.InputTokensDetails.TextTokens, - AudioTokens: ru.InputTokensDetails.AudioTokens, - ImageTokens: ru.InputTokensDetails.ImageTokens, - CachedReadTokens: ru.InputTokensDetails.CachedReadTokens, - CachedWriteTokens: ru.InputTokensDetails.CachedWriteTokens, + TextTokens: ru.InputTokensDetails.TextTokens, + AudioTokens: ru.InputTokensDetails.AudioTokens, + ImageTokens: ru.InputTokensDetails.ImageTokens, + CachedReadTokens: ru.InputTokensDetails.CachedReadTokens, + CachedWriteTokens: ru.InputTokensDetails.CachedWriteTokens, + CachedWriteTokenDetails: ru.InputTokensDetails.CachedWriteTokenDetails, } } if ru.OutputTokensDetails != nil { diff --git a/flake.lock b/flake.lock index 6eef0dc1c4f..346ad627a62 100644 --- a/flake.lock +++ b/flake.lock @@ -2,11 +2,11 @@ "nodes": { "nixpkgs": { "locked": { - "lastModified": 1776062742, - "narHash": "sha256-CYncVXVsUzYK+JZldSuK08ibXrAIJh+T22V13Z4ySS0=", + "lastModified": 1779887442, + "narHash": "sha256-eCUuOLWs77dwTqxIHTIZM4Wnxy+lirCLo6HjZr5dlgk=", "owner": "NixOS", "repo": "nixpkgs", - "rev": "1c742e001e98f5191a5586751e16311fe1481f61", + "rev": "a90757c3affe4befdc02025c1ae72df4f2b4be9e", "type": "github" }, "original": { diff --git a/plugins/otel/main.go b/plugins/otel/main.go index d300557ad17..005277e7dc3 100644 --- a/plugins/otel/main.go +++ b/plugins/otel/main.go @@ -8,6 +8,7 @@ import ( "strings" "github.com/bytedance/sonic" + bifrost "github.com/maximhq/bifrost/core" "github.com/maximhq/bifrost/core/schemas" "github.com/maximhq/bifrost/framework/modelcatalog" "go.opentelemetry.io/otel/attribute" @@ -421,9 +422,35 @@ func (p *OtelPlugin) PreLLMHook(_ *schemas.BifrostContext, req *schemas.BifrostR return req, nil, nil } -// PostLLMHook is a no-op - tracing is handled via the Inject method. -// The OTEL plugin receives completed traces from TracingMiddleware. -func (p *OtelPlugin) PostLLMHook(_ *schemas.BifrostContext, resp *schemas.BifrostResponse, bifrostErr *schemas.BifrostError) (*schemas.BifrostResponse, *schemas.BifrostError, error) { +// PostLLMHook records the cache-hit metric. Every other metric is derived from the +// completed trace in recordMetricsFromTrace, but semantic-cache hits short-circuit the +// request in a PreHook before any llm.call span exists, so the cache signal never reaches +// a span. We therefore read CacheDebug straight off the response here, mirroring how the +// Prometheus telemetry plugin and the Datadog plugin emit this metric. +// +// This is the ONLY place RecordCacheHit is called — do not also emit it from +// recordMetricsFromTrace, or cache hits will double-count. +func (p *OtelPlugin) PostLLMHook(ctx *schemas.BifrostContext, resp *schemas.BifrostResponse, bifrostErr *schemas.BifrostError) (*schemas.BifrostResponse, *schemas.BifrostError, error) { + if p.metricsExporter == nil || resp == nil { + return resp, bifrostErr, nil + } + extra := resp.GetExtraFields() + if extra == nil || extra.CacheDebug == nil || !extra.CacheDebug.CacheHit { + return resp, bifrostErr, nil + } + + cacheType := "unknown" + if extra.CacheDebug.HitType != nil && *extra.CacheDebug.HitType != "" { + cacheType = *extra.CacheDebug.HitType + } + + // Same dimensions as the trace-derived metrics (so the cache-hit counter shares labels + // with every other bifrost_* OTEL metric), but sourced from context — a short-circuited + // cache hit has no span to read. + attrs := append(buildContextAttrs(ctx, resp, bifrostErr), attribute.String("cache_type", cacheType)) + + p.metricsExporter.RecordCacheHit(ctx, attrs...) + return resp, bifrostErr, nil } @@ -515,6 +542,31 @@ func buildSpanAttrs(span *schemas.Span) []attribute.KeyValue { ) } +// buildContextAttrs builds the same metric dimension attrs as buildSpanAttrs, but sourced +// from the request context and response instead of a completed span. Used by hook-based +// metrics (e.g. cache hits) that fire without a provider-attempt span to read from. +func buildContextAttrs(ctx context.Context, resp *schemas.BifrostResponse, bifrostErr *schemas.BifrostError) []attribute.KeyValue { + requestType, provider, originalModel, resolvedModel := bifrost.GetResponseFields(resp, bifrostErr) + model := originalModel + if resolvedModel != "" { + model = resolvedModel + } + return BuildBifrostAttributes( + string(provider), + model, + string(requestType), + bifrost.GetStringFromContext(ctx, schemas.BifrostContextKeyGovernanceVirtualKeyID), + bifrost.GetStringFromContext(ctx, schemas.BifrostContextKeyGovernanceVirtualKeyName), + bifrost.GetStringFromContext(ctx, schemas.BifrostContextKeySelectedKeyID), + bifrost.GetStringFromContext(ctx, schemas.BifrostContextKeySelectedKeyName), + bifrost.GetIntFromContext(ctx, schemas.BifrostContextKeyFallbackIndex), + bifrost.GetStringFromContext(ctx, schemas.BifrostContextKeyGovernanceTeamID), + bifrost.GetStringFromContext(ctx, schemas.BifrostContextKeyGovernanceTeamName), + bifrost.GetStringFromContext(ctx, schemas.BifrostContextKeyGovernanceCustomerID), + bifrost.GetStringFromContext(ctx, schemas.BifrostContextKeyGovernanceCustomerName), + ) +} + // recordMetricsFromTrace extracts metrics data from a completed trace and records them // via the OTEL metrics exporter. This is called from Inject after trace emission. // @@ -597,6 +649,33 @@ func (p *OtelPlugin) recordMetricsFromTrace(ctx context.Context, trace *schemas. // Convert from nanoseconds to seconds if needed (check the unit) p.metricsExporter.RecordStreamFirstTokenLatency(ctx, ttft/1e9, otelAttrs...) } + + // Record provider-side prompt cache tokens (cache_read / cache_creation). Unlike the + // cache-hit counter, these ride real upstream calls, so the values are on the final + // attempt span just like input/output tokens. The read/write totals share unified + // span-attr keys across the chat and responses APIs; the 5m/1h breakdown uses + // API-family-specific keys that are mutually exclusive per request, so a fallback read + // covers both. + if n := getIntAttr(attrs, schemas.AttrUsageCacheReadInputTokens); n > 0 { + p.metricsExporter.RecordCacheReadInputTokens(ctx, int64(n), otelAttrs...) + } + if n := getIntAttr(attrs, schemas.AttrUsageCacheCreationInputTokens); n > 0 { + p.metricsExporter.RecordCacheWriteInputTokens(ctx, int64(n), otelAttrs...) + } + cacheWrite5m := getIntAttr(attrs, schemas.AttrPromptTokenDetailsCachedWrite5m) + if cacheWrite5m == 0 { + cacheWrite5m = getIntAttr(attrs, schemas.AttrInputTokenDetailsCachedWrite5m) + } + if cacheWrite5m > 0 { + p.metricsExporter.RecordCacheWriteInputTokens5m(ctx, int64(cacheWrite5m), otelAttrs...) + } + cacheWrite1h := getIntAttr(attrs, schemas.AttrPromptTokenDetailsCachedWrite1h) + if cacheWrite1h == 0 { + cacheWrite1h = getIntAttr(attrs, schemas.AttrInputTokenDetailsCachedWrite1h) + } + if cacheWrite1h > 0 { + p.metricsExporter.RecordCacheWriteInputTokens1h(ctx, int64(cacheWrite1h), otelAttrs...) + } } // Cleanup function for the OTEL plugin diff --git a/plugins/otel/metrics.go b/plugins/otel/metrics.go index d7359a0a7de..57e41bac119 100644 --- a/plugins/otel/metrics.go +++ b/plugins/otel/metrics.go @@ -46,6 +46,13 @@ type MetricsExporter struct { outputTokensTotal *syncInt64Counter cacheHitsTotal *syncInt64Counter + // Provider-side prompt cache token counters (distinct from cacheHitsTotal, which + // counts Bifrost's own semantic-cache hits). + cacheReadInputTokensTotal *syncInt64Counter + cacheWriteInputTokensTotal *syncInt64Counter + cacheWriteInputTokens5mTotal *syncInt64Counter + cacheWriteInputTokens1hTotal *syncInt64Counter + // Bifrost metrics - float counters (for cost) costTotal *syncFloat64Counter @@ -405,6 +412,34 @@ func (m *MetricsExporter) initMetrics() { meter: m.meter, } + m.cacheReadInputTokensTotal = &syncInt64Counter{ + name: "bifrost_cache_read_input_tokens_total", + desc: "Total provider-side prompt-cache read (cached) input tokens. Billed at a reduced rate by the provider", + unit: "{token}", + meter: m.meter, + } + + m.cacheWriteInputTokensTotal = &syncInt64Counter{ + name: "bifrost_cache_write_input_tokens_total", + desc: "Total provider-side prompt-cache creation (write) input tokens", + unit: "{token}", + meter: m.meter, + } + + m.cacheWriteInputTokens5mTotal = &syncInt64Counter{ + name: "bifrost_cache_write_input_tokens_5m_total", + desc: "Provider-side prompt-cache write input tokens with a 5-minute TTL (Anthropic only). Subset of bifrost_cache_write_input_tokens_total — do not sum with it", + unit: "{token}", + meter: m.meter, + } + + m.cacheWriteInputTokens1hTotal = &syncInt64Counter{ + name: "bifrost_cache_write_input_tokens_1h_total", + desc: "Provider-side prompt-cache write input tokens with a 1-hour TTL (Anthropic only). Subset of bifrost_cache_write_input_tokens_total — do not sum with it", + unit: "{token}", + meter: m.meter, + } + m.costTotal = &syncFloat64Counter{ name: "bifrost_cost_total", desc: "Total cost in USD for requests to upstream providers", @@ -513,6 +548,26 @@ func (m *MetricsExporter) RecordCacheHit(ctx context.Context, attrs ...attribute m.cacheHitsTotal.Add(ctx, 1, metric.WithAttributes(attrs...)) } +// RecordCacheReadInputTokens records provider-side prompt-cache read (cached) input tokens. +func (m *MetricsExporter) RecordCacheReadInputTokens(ctx context.Context, count int64, attrs ...attribute.KeyValue) { + m.cacheReadInputTokensTotal.Add(ctx, count, metric.WithAttributes(attrs...)) +} + +// RecordCacheWriteInputTokens records provider-side prompt-cache creation (write) input tokens. +func (m *MetricsExporter) RecordCacheWriteInputTokens(ctx context.Context, count int64, attrs ...attribute.KeyValue) { + m.cacheWriteInputTokensTotal.Add(ctx, count, metric.WithAttributes(attrs...)) +} + +// RecordCacheWriteInputTokens5m records the 5-minute-TTL subset of cache-write input tokens. +func (m *MetricsExporter) RecordCacheWriteInputTokens5m(ctx context.Context, count int64, attrs ...attribute.KeyValue) { + m.cacheWriteInputTokens5mTotal.Add(ctx, count, metric.WithAttributes(attrs...)) +} + +// RecordCacheWriteInputTokens1h records the 1-hour-TTL subset of cache-write input tokens. +func (m *MetricsExporter) RecordCacheWriteInputTokens1h(ctx context.Context, count int64, attrs ...attribute.KeyValue) { + m.cacheWriteInputTokens1hTotal.Add(ctx, count, metric.WithAttributes(attrs...)) +} + // RecordCost records cost metric func (m *MetricsExporter) RecordCost(ctx context.Context, cost float64, attrs ...attribute.KeyValue) { m.costTotal.Add(ctx, cost, metric.WithAttributes(attrs...)) diff --git a/plugins/telemetry/main.go b/plugins/telemetry/main.go index 587fdc02aa3..93f25fc326c 100644 --- a/plugins/telemetry/main.go +++ b/plugins/telemetry/main.go @@ -162,6 +162,10 @@ type PrometheusPlugin struct { InputTokensTotal *prometheus.CounterVec OutputTokensTotal *prometheus.CounterVec CacheHitsTotal *prometheus.CounterVec + CacheReadInputTokensTotal *prometheus.CounterVec + CacheWriteInputTokensTotal *prometheus.CounterVec + CacheWriteInputTokens5mTotal *prometheus.CounterVec + CacheWriteInputTokens1hTotal *prometheus.CounterVec CostTotal *prometheus.CounterVec StreamInterTokenLatencySeconds *prometheus.HistogramVec StreamFirstTokenLatencySeconds *prometheus.HistogramVec @@ -380,6 +384,40 @@ func Init(config *Config, pricingManager *modelcatalog.ModelCatalog, logger sche append(append(defaultBifrostLabels, "cache_type"), filteredCustomLabels...), ) + // Provider-side prompt cache tokens (Anthropic/OpenAI/Gemini prompt caching). Distinct + // from bifrost_cache_hits_total, which counts Bifrost's own semantic-cache hits. + bifrostCacheReadInputTokensTotal := factory.NewCounterVec( + prometheus.CounterOpts{ + Name: "bifrost_cache_read_input_tokens_total", + Help: "Total provider-side prompt-cache read (cached) input tokens. Billed at a reduced rate by the provider.", + }, + append(defaultBifrostLabels, filteredCustomLabels...), + ) + + bifrostCacheWriteInputTokensTotal := factory.NewCounterVec( + prometheus.CounterOpts{ + Name: "bifrost_cache_write_input_tokens_total", + Help: "Total provider-side prompt-cache creation (write) input tokens.", + }, + append(defaultBifrostLabels, filteredCustomLabels...), + ) + + bifrostCacheWriteInputTokens5mTotal := factory.NewCounterVec( + prometheus.CounterOpts{ + Name: "bifrost_cache_write_input_tokens_5m_total", + Help: "Provider-side prompt-cache write input tokens with a 5-minute TTL (Anthropic only). Subset of bifrost_cache_write_input_tokens_total — do not sum with it.", + }, + append(defaultBifrostLabels, filteredCustomLabels...), + ) + + bifrostCacheWriteInputTokens1hTotal := factory.NewCounterVec( + prometheus.CounterOpts{ + Name: "bifrost_cache_write_input_tokens_1h_total", + Help: "Provider-side prompt-cache write input tokens with a 1-hour TTL (Anthropic only). Subset of bifrost_cache_write_input_tokens_total — do not sum with it.", + }, + append(defaultBifrostLabels, filteredCustomLabels...), + ) + bifrostCostTotal := factory.NewCounterVec( prometheus.CounterOpts{ Name: "bifrost_cost_total", @@ -460,6 +498,10 @@ func Init(config *Config, pricingManager *modelcatalog.ModelCatalog, logger sche InputTokensTotal: bifrostInputTokensTotal, OutputTokensTotal: bifrostOutputTokensTotal, CacheHitsTotal: bifrostCacheHitsTotal, + CacheReadInputTokensTotal: bifrostCacheReadInputTokensTotal, + CacheWriteInputTokensTotal: bifrostCacheWriteInputTokensTotal, + CacheWriteInputTokens5mTotal: bifrostCacheWriteInputTokens5mTotal, + CacheWriteInputTokens1hTotal: bifrostCacheWriteInputTokens1hTotal, CostTotal: bifrostCostTotal, StreamInterTokenLatencySeconds: bifrostStreamInterTokenLatencySeconds, StreamFirstTokenLatencySeconds: bifrostStreamFirstTokenLatencySeconds, @@ -577,6 +619,41 @@ func (p *PrometheusPlugin) PreLLMHook(ctx *schemas.BifrostContext, req *schemas. return req, nil, nil } +// extractProviderCacheTokens returns provider-side prompt-cache token counts from a +// response's usage: cache-read (cached) input tokens, cache-write (creation) input tokens, +// and the Anthropic-only 5m/1h TTL breakdown of the write total. Chat/text-completion carry +// these on Usage.PromptTokensDetails; the Responses API carries them on +// Usage.InputTokensDetails. Mirrors the response-type switch used for input/output tokens. +func extractProviderCacheTokens(result *schemas.BifrostResponse) (read, write, write5m, write1h int) { + var promptDetails *schemas.ChatPromptTokensDetails + var inputDetails *schemas.ResponsesResponseInputTokens + + switch { + case result.TextCompletionResponse != nil && result.TextCompletionResponse.Usage != nil: + promptDetails = result.TextCompletionResponse.Usage.PromptTokensDetails + case result.ChatResponse != nil && result.ChatResponse.Usage != nil: + promptDetails = result.ChatResponse.Usage.PromptTokensDetails + case result.ResponsesResponse != nil && result.ResponsesResponse.Usage != nil: + inputDetails = result.ResponsesResponse.Usage.InputTokensDetails + case result.ResponsesStreamResponse != nil && result.ResponsesStreamResponse.Response != nil && result.ResponsesStreamResponse.Response.Usage != nil: + inputDetails = result.ResponsesStreamResponse.Response.Usage.InputTokensDetails + } + + switch { + case promptDetails != nil: + read, write = promptDetails.CachedReadTokens, promptDetails.CachedWriteTokens + if d := promptDetails.CachedWriteTokenDetails; d != nil { + write5m, write1h = d.CachedWriteTokens5m, d.CachedWriteTokens1h + } + case inputDetails != nil: + read, write = inputDetails.CachedReadTokens, inputDetails.CachedWriteTokens + if d := inputDetails.CachedWriteTokenDetails; d != nil { + write5m, write1h = d.CachedWriteTokens5m, d.CachedWriteTokens1h + } + } + return +} + // PostLLMHook calculates duration and records upstream metrics for successful requests. // It records: // - Request latency @@ -808,6 +885,23 @@ func (p *PrometheusPlugin) PostLLMHook(ctx *schemas.BifrostContext, result *sche p.InputTokensTotal.WithLabelValues(promLabelValues...).Add(float64(inputTokens)) p.OutputTokensTotal.WithLabelValues(promLabelValues...).Add(float64(outputTokens)) + // Record provider-side prompt cache tokens (Anthropic/OpenAI/Gemini prompt + // caching). Distinct from the cache-hit counter below, which tracks Bifrost's + // own semantic cache. 5m/1h are an Anthropic-only TTL breakdown of the write total. + cacheRead, cacheWrite, cacheWrite5m, cacheWrite1h := extractProviderCacheTokens(result) + if cacheRead > 0 { + p.CacheReadInputTokensTotal.WithLabelValues(promLabelValues...).Add(float64(cacheRead)) + } + if cacheWrite > 0 { + p.CacheWriteInputTokensTotal.WithLabelValues(promLabelValues...).Add(float64(cacheWrite)) + } + if cacheWrite5m > 0 { + p.CacheWriteInputTokens5mTotal.WithLabelValues(promLabelValues...).Add(float64(cacheWrite5m)) + } + if cacheWrite1h > 0 { + p.CacheWriteInputTokens1hTotal.WithLabelValues(promLabelValues...).Add(float64(cacheWrite1h)) + } + // Record cache hits with cache type extraFields := result.GetExtraFields() if extraFields.CacheDebug != nil && extraFields.CacheDebug.CacheHit {