[management, proxy] Management-owned LLM pricing: file-backed defaults + (#6965)

This commit is contained in:
Misha Bragin
2026-07-31 20:52:56 +02:00
committed by GitHub
parent feecb993f4
commit 0780a806f2
47 changed files with 3196 additions and 1540 deletions
+116 -371
View File
@@ -1,102 +1,30 @@
// Package pricing implements the embedded-default + override pricing table
// shared by middleware that converts LLM token usage into a USD cost
// estimate. The table is hot-reloadable from a basename under the proxy
// data directory; missing override files keep the embedded defaults so
// cost annotation works without operator action.
// Package pricing implements the pricing table and cost formula the
// cost_meter middleware uses to convert LLM token usage into a USD cost
// estimate. The table's content arrives from the management server inside
// cost_meter's middleware config (synthesized from the catalog plus the
// operator's stored per-provider prices) — the proxy carries no embedded
// price list. Price updates ride the ordinary mapping push: a chain
// rebuild constructs a fresh table, so there is nothing to reload.
package pricing
import (
"bytes"
"context"
_ "embed"
"errors"
"fmt"
"io"
"io/fs"
"math"
"path/filepath"
"regexp"
"strings"
"sync"
"sync/atomic"
"time"
log "github.com/sirupsen/logrus"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/metric"
"gopkg.in/yaml.v3"
)
//go:embed defaults_pricing.yaml
var defaultPricingYAML []byte
var (
defaultTableOnce sync.Once
defaultTablePtr *Table
)
// DefaultTable returns the pricing table embedded in the binary. The result
// is parsed once and shared; callers must not mutate the returned value.
// Cost annotation works without any operator action because every loader
// starts with this table.
func DefaultTable() *Table {
defaultTableOnce.Do(func() {
t, err := parsePricingBytes(defaultPricingYAML)
if err != nil {
panic(fmt.Sprintf("llmobs: embedded default pricing failed to parse: %v", err))
}
defaultTablePtr = t
})
return defaultTablePtr
}
// mergeOver returns a new Table containing every entry from base, with any
// matching entry from overlay replacing the base value. Either argument may
// be nil. Result is a fresh allocation so callers can mutate / Store safely.
func mergeOver(base, overlay *Table) *Table {
if overlay == nil || len(overlay.entries) == 0 {
return base
}
if base == nil || len(base.entries) == 0 {
return overlay
}
out := make(map[string]map[string]Entry, len(base.entries))
for provider, models := range base.entries {
inner := make(map[string]Entry, len(models))
for model, e := range models {
inner[model] = e
}
out[provider] = inner
}
for provider, models := range overlay.entries {
inner, ok := out[provider]
if !ok {
inner = make(map[string]Entry, len(models))
out[provider] = inner
}
for model, e := range models {
inner[model] = e
}
}
return &Table{entries: out}
}
// Entry is a single model's input and output pricing, expressed in USD per
// 1000 tokens.
//
// CachedInputPer1K applies to OpenAI's cached prompt tokens, which are a
// subset of input_tokens — when set, the cached portion is billed at this
// rate and the non-cached remainder at InputPer1K. Zero means "no discount
// configured", and cached tokens are billed at InputPer1K (matches current
// behaviour where cached counts weren't extracted at all).
// configured", and cached tokens are billed at InputPer1K.
//
// CacheReadPer1K and CacheCreationPer1K apply to Anthropic's two prompt-
// cache fields, which are additive to input_tokens: cache_read is the
// cheaper read-from-cache rate, cache_creation is the more expensive
// write-to-cache rate. Zero means "no rate configured" and the
// corresponding token bucket is billed at InputPer1K. This is more
// accurate than today's behaviour, where Anthropic's cache tokens are
// ignored and not charged at all.
// corresponding token bucket is billed at InputPer1K.
type Entry struct {
InputPer1K float64
OutputPer1K float64
@@ -105,33 +33,102 @@ type Entry struct {
CacheCreationPer1K float64
}
// Table is a provider-to-model pricing lookup. Instances are immutable once
// built and are swapped atomically by Loader.
// EntryJSON is the wire shape of a pricing entry inside cost_meter's
// middleware config. Field names are the management→proxy contract; the
// management synthesizer marshals the same names (its pricing.Entry).
type EntryJSON struct {
InputPer1K float64 `json:"input_per_1k"`
OutputPer1K float64 `json:"output_per_1k"`
CachedInputPer1K float64 `json:"cached_input_per_1k"`
CacheReadPer1K float64 `json:"cache_read_per_1k"`
CacheCreationPer1K float64 `json:"cache_creation_per_1k"`
}
// Table is a provider-surface-to-model pricing lookup. Instances are
// immutable once built; a mapping update builds a whole new middleware
// instance (and with it a new table) rather than mutating this one.
type Table struct {
entries map[string]map[string]Entry
}
// NewEntries validates and converts a wire-shape map (surface-or-record ->
// model -> rates) into the internal representation. Every rate must be a
// finite, non-negative USD amount; a violation is returned as an error so
// a corrupt config fails the chain build loudly instead of mispricing.
// Management validates the same constraints at its API boundary, so this
// is defense-in-depth. Nil input yields an empty (never-matching) map.
func NewEntries(raw map[string]map[string]EntryJSON) (map[string]map[string]Entry, error) {
out := make(map[string]map[string]Entry, len(raw))
for outer, models := range raw {
inner := make(map[string]Entry, len(models))
for model, e := range models {
for field, v := range map[string]float64{
"input_per_1k": e.InputPer1K,
"output_per_1k": e.OutputPer1K,
"cached_input_per_1k": e.CachedInputPer1K,
"cache_read_per_1k": e.CacheReadPer1K,
"cache_creation_per_1k": e.CacheCreationPer1K,
} {
if v < 0 || math.IsNaN(v) || math.IsInf(v, 0) {
return nil, fmt.Errorf("pricing %s/%s: %s must be a finite, non-negative rate, got %v", outer, model, field, v)
}
}
// EntryJSON and Entry are field-identical (tags aside), so a
// direct conversion carries all five rates.
inner[model] = Entry(e)
}
out[outer] = inner
}
return out, nil
}
// NewTable builds an immutable Table from the wire-shape defaults map.
// See NewEntries for validation semantics.
func NewTable(raw map[string]map[string]EntryJSON) (*Table, error) {
entries, err := NewEntries(raw)
if err != nil {
return nil, err
}
return &Table{entries: entries}, nil
}
// Lookup returns the entry for the given provider surface and model.
func (t *Table) Lookup(provider, model string) (Entry, bool) {
if t == nil {
return Entry{}, false
}
byModel, ok := t.entries[provider]
if !ok {
return Entry{}, false
}
e, ok := byModel[model]
return e, ok
}
// Has reports whether the provider/model pair is present in the table.
func (t *Table) Has(provider, model string) bool {
_, ok := t.Lookup(provider, model)
return ok
}
// Cost returns the estimated USD cost for the given token counts. ok is
// false when the provider or model is not present in the table; the caller
// can still emit token metrics with a model=unknown label.
//
// Provider-shape semantics for cached / cache-creation counts:
//
// - OpenAI: cachedInput is a SUBSET of inTokens. The cached portion is
// billed at CachedInputPer1K (or InputPer1K when no override), and the
// non-cached remainder of inTokens at InputPer1K. cacheCreation is
// ignored (OpenAI has no analogue).
// - Anthropic: cachedInput (cache_read) and cacheCreation are ADDITIVE to
// inTokens. The three buckets are billed at CacheReadPer1K,
// CacheCreationPer1K, and InputPer1K respectively, each falling back
// to InputPer1K when the corresponding rate is zero.
// - Other providers: cached and cacheCreation are ignored; cost is
// inTokens*InputPer1K + outTokens*OutputPer1K.
func (t *Table) Cost(provider, model string, inTokens, outTokens, cachedInput, cacheCreation int64) (float64, bool) {
c, ok := t.Costs(provider, model, inTokens, outTokens, cachedInput, cacheCreation)
return c.TotalUSD, ok
}
// Costs returns the estimated USD cost split for the given token counts.
// The provider surface selects the cache formula; see EntryCosts.
func (t *Table) Costs(provider, model string, inTokens, outTokens, cachedInput, cacheCreation int64) (Costs, bool) {
entry, ok := t.Lookup(provider, model)
if !ok {
return Costs{}, false
}
return EntryCosts(entry, provider, inTokens, outTokens, cachedInput, cacheCreation), true
}
// Costs is a per-request cost split. The four per-bucket fields are the base
// of the breakdown — one per token bucket the provider bills separately — and
// the two aggregates are derived from them:
@@ -165,9 +162,25 @@ func newCosts(input, cachedInput, cacheCreation, output float64) Costs {
}
}
// Costs returns the estimated USD cost split for the given token counts, with
// the same semantics as Cost.
func (t *Table) Costs(provider, model string, inTokens, outTokens, cachedInput, cacheCreation int64) (Costs, bool) {
// EntryCosts computes the USD cost split for the given entry and token
// counts. The surface (the llm.provider value the request parser stamps)
// selects the cache formula; the entry may come from the surface-keyed
// defaults table or from a per-provider-record override — the math is
// identical either way.
//
// Provider-shape semantics for cached / cache-creation counts:
//
// - "openai": cachedInput is a SUBSET of inTokens. The cached portion is
// billed at CachedInputPer1K (or InputPer1K when no override), and the
// non-cached remainder of inTokens at InputPer1K. cacheCreation is
// ignored (OpenAI has no analogue).
// - "anthropic", "bedrock": cachedInput (cache_read) and cacheCreation are
// ADDITIVE to inTokens. The three buckets are billed at CacheReadPer1K,
// CacheCreationPer1K, and InputPer1K respectively, each falling back
// to InputPer1K when the corresponding rate is zero.
// - Other surfaces: cached and cacheCreation are ignored; cost is
// inTokens*InputPer1K + outTokens*OutputPer1K.
func EntryCosts(entry Entry, surface string, inTokens, outTokens, cachedInput, cacheCreation int64) Costs {
// Clamp negatives to zero before any pricing math so a malformed
// upstream count can never produce a negative cost.
if inTokens < 0 {
@@ -182,19 +195,8 @@ func (t *Table) Costs(provider, model string, inTokens, outTokens, cachedInput,
if cacheCreation < 0 {
cacheCreation = 0
}
if t == nil {
return Costs{}, false
}
byModel, ok := t.entries[provider]
if !ok {
return Costs{}, false
}
entry, ok := byModel[model]
if !ok {
return Costs{}, false
}
output := (float64(outTokens) / 1000.0) * entry.OutputPer1K
switch provider {
switch surface {
case "openai":
// cachedInput is a subset of inTokens; clamp so a malformed
// upstream (cached > total) can't produce a negative remainder.
@@ -208,7 +210,7 @@ func (t *Table) Costs(provider, model string, inTokens, outTokens, cachedInput,
}
nonCached := float64(inTokens-clamped) / 1000.0 * entry.InputPer1K
cached := float64(clamped) / 1000.0 * cachedRate
return newCosts(nonCached, cached, 0, output), true
return newCosts(nonCached, cached, 0, output)
case "anthropic", "bedrock":
// Bedrock-Anthropic returns the same additive cache buckets as
// first-party Anthropic; non-Anthropic Bedrock models simply report
@@ -224,266 +226,9 @@ func (t *Table) Costs(provider, model string, inTokens, outTokens, cachedInput,
input := float64(inTokens) / 1000.0 * entry.InputPer1K
read := float64(cachedInput) / 1000.0 * readRate
create := float64(cacheCreation) / 1000.0 * createRate
return newCosts(input, read, create, output), true
return newCosts(input, read, create, output)
default:
input := float64(inTokens) / 1000.0 * entry.InputPer1K
return newCosts(input, 0, 0, output), true
return newCosts(input, 0, 0, output)
}
}
// Has reports whether the provider/model pair is present in the table.
func (t *Table) Has(provider, model string) bool {
if t == nil {
return false
}
byModel, ok := t.entries[provider]
if !ok {
return false
}
_, ok = byModel[model]
return ok
}
// pricingFile mirrors the on-disk YAML schema. Keys are provider names; the
// nested map keys are model names.
type pricingFile map[string]map[string]struct {
InputPer1K float64 `yaml:"input_per_1k"`
OutputPer1K float64 `yaml:"output_per_1k"`
CachedInputPer1K float64 `yaml:"cached_input_per_1k"`
CacheReadPer1K float64 `yaml:"cache_read_per_1k"`
CacheCreationPer1K float64 `yaml:"cache_creation_per_1k"`
}
const (
// ReloadInterval is the mtime-poll cadence for the background reloader.
ReloadInterval = 30 * time.Second
// errorBackoff bounds how often the loader logs a repeated parse error.
errorBackoff = 5 * time.Minute
)
var basenameRegex = regexp.MustCompile(`^[a-zA-Z0-9._-]+$`)
// Loader is a confined, hot-reloadable pricing table reader. Construction
// must succeed against the target file; subsequent reload failures keep the
// previously-loaded table so callers never observe a blank price list.
type Loader struct {
baseDir string
fullPath string
pluginID string
table atomic.Pointer[Table]
mtime atomic.Int64
failures metric.Int64Counter
interval time.Duration
}
// NewLoader returns a pricing loader that overlays an optional file-based
// table on top of the embedded defaults. Missing override file, baseDir, or
// relPath is not an error: the loader keeps the embedded defaults so cost
// metadata is still emitted for known models.
//
// Errors:
// - bad basename, traversal segment, or absolute relPath are rejected so a
// misconfigured target surfaces immediately.
// - permission errors and YAML parse errors keep the defaults but log a
// warning; cost annotation does not silently break.
//
// failures is optional; pass nil in tests that do not care about
// reload-failure telemetry.
func NewLoader(baseDir, relPath, pluginID string, failures metric.Int64Counter) (*Loader, error) {
defaults := DefaultTable()
l := &Loader{
baseDir: baseDir,
pluginID: pluginID,
failures: failures,
}
l.table.Store(defaults)
if strings.TrimSpace(baseDir) == "" || strings.TrimSpace(relPath) == "" {
return l, nil
}
full, err := resolveMiddlewareDataPath(baseDir, relPath)
if err != nil {
return nil, err
}
l.fullPath = full
overlay, mtime, err := loadPricing(full)
if err != nil {
if errors.Is(err, fs.ErrNotExist) {
// Override file is optional. Defaults already stored.
return l, nil
}
// Symlink rejection, oversize file, parse failure, permission errors
// — surface so a misconfigured operator sees the problem instead of
// silently running with stale defaults.
return nil, fmt.Errorf("load pricing %s: %w", full, err)
}
l.table.Store(mergeOver(defaults, overlay))
l.mtime.Store(mtime.UnixNano())
return l, nil
}
// Get returns the current pricing table. The returned pointer is immutable;
// callers must not mutate its contents.
func (l *Loader) Get() *Table {
if l == nil {
return nil
}
return l.table.Load()
}
// WatchesFile reports whether this loader is bound to an override file on
// disk. False for defaults-only loaders (no operator override given).
// Callers use this to decide whether to spawn the mtime-poll goroutine.
func (l *Loader) WatchesFile() bool {
if l == nil {
return false
}
return l.fullPath != ""
}
// SetReloadInterval overrides the mtime-poll cadence used by Reload. Calls
// after Reload has started have no effect on the running loop. Intended for
// tests; production code uses the default ReloadInterval.
func (l *Loader) SetReloadInterval(d time.Duration) {
if l == nil || d <= 0 {
return
}
l.interval = d
}
// Reload runs a polling loop that checks the pricing file mtime every
// ReloadInterval (or the value passed to SetReloadInterval). Returns when
// ctx is cancelled.
func (l *Loader) Reload(ctx context.Context) {
if l == nil {
return
}
interval := l.interval
if interval <= 0 {
interval = ReloadInterval
}
t := time.NewTicker(interval)
defer t.Stop()
var lastErrAt time.Time
for {
select {
case <-ctx.Done():
return
case <-t.C:
if err := l.reload(); err != nil {
if l.failures != nil {
l.failures.Add(ctx, 1, metric.WithAttributes(
attribute.String("plugin", l.pluginID),
))
}
now := time.Now()
if now.Sub(lastErrAt) >= errorBackoff {
log.Warnf("llmobs: pricing reload failed for %s: %v", l.fullPath, err)
lastErrAt = now
}
}
}
}
}
// reload performs a single-shot mtime check and reload. The reloaded
// override file is merged on top of the embedded defaults; missing override
// (e.g. operator deleted the file) is not an error and reverts to defaults.
func (l *Loader) reload() error {
if l.fullPath == "" {
// Defaults-only loader; nothing on disk to reload.
return nil
}
mtime, err := statMtime(l.fullPath)
if err != nil {
if errors.Is(err, fs.ErrNotExist) {
// File was removed since startup. Drop back to defaults and
// reset mtime so a future re-creation triggers a reload.
l.table.Store(DefaultTable())
l.mtime.Store(0)
return nil
}
return err
}
if mtime.UnixNano() == l.mtime.Load() {
return nil
}
overlay, newMtime, err := loadPricing(l.fullPath)
if err != nil {
return err
}
l.table.Store(mergeOver(DefaultTable(), overlay))
l.mtime.Store(newMtime.UnixNano())
return nil
}
// resolveMiddlewareDataPath validates relPath is a safe basename and resolves
// it under baseDir. An additional cleaned-prefix check guards against
// CVE-style edge cases where Join is used with trailing path segments.
func resolveMiddlewareDataPath(baseDir, relPath string) (string, error) {
if strings.TrimSpace(baseDir) == "" {
return "", errors.New("middleware-data-dir is not configured")
}
if relPath == "" {
return "", errors.New("pricing path is empty")
}
if !basenameRegex.MatchString(relPath) {
return "", fmt.Errorf("pricing path %q is not a safe basename", relPath)
}
if filepath.IsAbs(relPath) {
return "", fmt.Errorf("pricing path %q must be a basename, not absolute", relPath)
}
cleanBase, err := filepath.Abs(filepath.Clean(baseDir))
if err != nil {
return "", fmt.Errorf("resolve middleware-data-dir: %w", err)
}
full := filepath.Join(cleanBase, relPath)
cleanedFull := filepath.Clean(full)
if !strings.HasPrefix(cleanedFull, cleanBase+string(filepath.Separator)) && cleanedFull != cleanBase {
return "", fmt.Errorf("pricing path %q escapes middleware-data-dir", relPath)
}
return cleanedFull, nil
}
func parsePricingBytes(data []byte) (*Table, error) {
dec := yaml.NewDecoder(bytes.NewReader(data))
dec.KnownFields(true)
var raw pricingFile
if err := dec.Decode(&raw); err != nil && !errors.Is(err, io.EOF) {
return nil, fmt.Errorf("decode pricing yaml: %w", err)
}
out := make(map[string]map[string]Entry, len(raw))
for provider, models := range raw {
inner := make(map[string]Entry, len(models))
for model, entry := range models {
for field, v := range map[string]float64{
"input_per_1k": entry.InputPer1K,
"output_per_1k": entry.OutputPer1K,
"cached_input_per_1k": entry.CachedInputPer1K,
"cache_read_per_1k": entry.CacheReadPer1K,
"cache_creation_per_1k": entry.CacheCreationPer1K,
} {
if v < 0 || math.IsNaN(v) || math.IsInf(v, 0) {
return nil, fmt.Errorf("pricing %s/%s: %s must be a finite, non-negative rate, got %v", provider, model, field, v)
}
}
inner[model] = Entry{
InputPer1K: entry.InputPer1K,
OutputPer1K: entry.OutputPer1K,
CachedInputPer1K: entry.CachedInputPer1K,
CacheReadPer1K: entry.CacheReadPer1K,
CacheCreationPer1K: entry.CacheCreationPer1K,
}
}
out[provider] = inner
}
return &Table{entries: out}, nil
}