usage

package
v6.10.9-aug.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 14, 2026 License: MIT Imports: 29 Imported by: 0

Documentation

Overview

Package usage provides usage tracking and logging functionality for the CLI Proxy API server. It includes plugins for monitoring API usage, token consumption, and other metrics to help with observability and billing purposes.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ComputeFingerprint 

func ComputeFingerprint(d FlatDetail) [20]byte

ComputeFingerprint produces a 20-byte SHA-1 over the canonical dedupKey payload. Field order MUST stay aligned with dedupKey() in logger_plugin.go so the same logical event arriving via v2 detail / v1 nested / frontend CSV / frontend JSON / /usage/export collapses onto one row through the events.fingerprint UNIQUE constraint.

Provider compatibility: when Provider is empty (legacy JSON imports, where the field never existed on disk) the payload format reverts to the v1 11- field layout so historical rows reproduce the same SHA-1 the JSON era produced. When Provider is set (live writes from the v2 LoggerPlugin) the 12-field layout is used so the same email speaking to two providers no longer dedupes to a single row.

SHA-1 is used as a fast 160-bit hash, not for security; collision risk for a well-formed dedup payload is negligible (birthday bound ≫ realistic event counts) and any duplicate would harmlessly produce INSERT OR IGNORE no-ops.

func MigrateDataDirIfNeeded 

func MigrateDataDirIfNeeded(configFilePath, newBaseDir string) error

MigrateDataDirIfNeeded moves an existing usage-data directory from any legacy location to the resolved new location when WRITABLE_PATH or install layout changed between restarts. Cross-filesystem moves fall back to copy + remove. Skips when paths match or when the new location already has summary.json / events.db.

func MigrateLegacyToSQLite 

func MigrateLegacyToSQLite(ctx context.Context, store *Store, priceFn PriceFunc, baseDir string) error

MigrateLegacyToSQLite imports every JSON file the v1/v2 era left behind in baseDir into the store. Old JSON files are left in place — events.db is the authoritative data source after migration, and meta.migrated_from_json guarantees subsequent restarts skip the rescan.

Idempotent: a successful run records meta.migrated_from_json, and subsequent invocations short-circuit immediately without scanning disk.

On parse failure the migration aborts with the failing file's path attached, the meta row is NOT written, and the caller can decide whether to retry or surface the failure.

func ResolveDataDir 

func ResolveDataDir(configured, configFilePath string) string

ResolveDataDir picks the usage data directory: configured > WRITABLE_PATH/usage-data > config dir/usage-data.

func SetGlobalPersister 

func SetGlobalPersister(p *Persister)

SetGlobalPersister registers the active Persister for the LoggerPlugin to use.

func SetStatisticsEnabled 

func SetStatisticsEnabled(enabled bool)

SetStatisticsEnabled toggles whether in-memory statistics are recorded.

Types

type ArchivedDetail 

type ArchivedDetail struct {
	APIName   string
	ModelName string
	Detail    requestDetail
}

ArchivedDetail pairs a trimmed requestDetail with its API and model names for archival.

type EventFilters 

type EventFilters struct {
	Model  string
	Source string
	APIKey string
	Status string // "success", "failure", or "" (all)
	Search string
}

EventFilters specifies optional dimensions the dashboard layers on top of a time-range query. Multi-value filters are expressed as comma-separated strings to match the on-the-wire form sent by the management UI.

type FlatDetail 

type FlatDetail struct {
	Timestamp time.Time  `json:"timestamp"`
	Model     string     `json:"model"`
	Source    string     `json:"source"`     // credential file name
	AuthIndex string     `json:"auth_index"` // alternate credential identifier
	APIKey    string     `json:"api_key"`    // CPA access key used by the client
	Provider  string     `json:"provider"`   // upstream vendor (claude/codex/gemini/...)
	Tokens    tokenStats `json:"tokens"`
	Failed    bool       `json:"failed"`
}

FlatDetail is the canonical denormalised request record passed between the LoggerPlugin, the persister, and the SQLite store. Each field maps to a dim_* table or directly to the events row. Source and AuthIndex are kept separate so /usage/events can still answer "which credential file" while dim_credential collapses them into one user-visible label.

Provider was added in schema v2 to disambiguate credentials that share a source string across different upstream vendors (Claude OAuth and Codex OAuth using the same Google email being the canonical example). Empty string means "unknown / pre-v2" — fingerprint() and the writers treat that as identical to the historical NULL so legacy rows imported from the JSON era keep their original deduplication identity.

func ParseAny 

func ParseAny(data []byte) ([]FlatDetail, string, error)

ParseAny detects the format and returns the flattened FlatDetail slice. Returns the matched parser type for diagnostics in the caller's logs.

type ImportHook 

type ImportHook func(detail FlatDetail)

ImportHook is called for each successfully imported detail with the resolved fields.

type ImportResult 

type ImportResult struct {
	Added         int64
	Skipped       int64
	Format        string
	EarliestDayNs int64
	LatestDayNs   int64
	HasDayRange   bool
}

ImportResult mirrors RequestStatistics.MergeSnapshot's MergeResult so the HTTP handler can keep the same response shape.

func ImportBytes 

func ImportBytes(ctx context.Context, store *Store, priceFn PriceFunc, data []byte) (ImportResult, error)

ImportBytes is the synchronous entry point used by HTTP handlers. It parses the payload, then imports every detail in a single transaction so the caller sees consistent counts on success.

func ImportFile 

func ImportFile(ctx context.Context, store *Store, priceFn PriceFunc, path string) (ImportResult, error)

ImportFile reads, parses and imports a single file. Used by migrate_legacy and could be exposed via a future CLI subcommand.

type MergeResult 

type MergeResult struct {
	Added   int64 `json:"added"`
	Skipped int64 `json:"skipped"`
}

type MigrationStatus 

type MigrationStatus struct {
	From string
	At   time.Time
}

MigrationStatus reports the persisted migration marker. From=="" means no migration has been recorded yet.

func CheckMigrationStatus 

func CheckMigrationStatus(ctx context.Context, store *Store) (MigrationStatus, error)

CheckMigrationStatus reads the meta row that flags a completed JSON import.

type Parser 

type Parser interface {
	Detect(data []byte) bool
	Parse(data []byte) ([]FlatDetail, error)
}

Parser describes one importable on-disk format. ParseAny tries Detect on each parser in priority order; the first match wins.

type Persister 

type Persister struct {
	// contains filtered or unexported fields
}

Persister is the SQLite-backed coordinator for usage data. It owns the Store + Writer + rollup goroutine, exposes a stable surface to the rest of the codebase, and hides the schema / WAL / dim_table details. Public method signatures match the JSON-era persister 1:1 so handler / SDK / test code keeps compiling unchanged.

func NewPersister 

func NewPersister(baseDir string, retention config.UsageRetention) *Persister

NewPersister constructs a Persister rooted at baseDir. A zero-valued Days is filled with the default (-1, disabled).

func (*Persister) BaseDir 

func (p *Persister) BaseDir() string

BaseDir returns the usage data directory.

func (*Persister) DBSize 

func (p *Persister) DBSize() int64

DBSize returns the on-disk size of events.db plus its WAL/SHM sidecars. Returns 0 when the store is not yet open or the file does not exist.

func (*Persister) DBSizeStatus 

func (p *Persister) DBSizeStatus() (
	sizeBytes int64,
	maxBytes int64,
	warningThresholdPct int,
	warning bool,
	capped bool,
)

func (*Persister) HasPricing 

func (p *Persister) HasPricing() bool

HasPricing reports whether a price lookup function has been configured.

func (*Persister) Import 

func (p *Persister) Import(data []byte) (ImportResult, error)

Import reads a payload (any of the supported formats), inserts the events, and returns counts. Used by /usage/import.

func (*Persister) IsReady 

func (p *Persister) IsReady() bool

IsReady reports whether Start successfully opened the SQLite store. Callers (sdk/cliproxy/service.go) should branch on this immediately after Start: when not ready, they MUST call SetGlobalPersister(nil) so the LoggerPlugin's fallback to in-memory RequestStatistics kicks in. Without that, Record() silently drops every event behind the nil-writer guard.

func (*Persister) MigrationStatus 

func (p *Persister) MigrationStatus(ctx context.Context) (MigrationStatus, error)

MigrationStatus returns the persisted legacy JSON migration marker.

func (*Persister) Query 

func (p *Persister) Query(
	from, to time.Time,
	filters EventFilters,
	granularity string,
	includeGroups bool,
) (QueryResult, error)

Query is the read-side entry point used by /usage/summary.

func (*Persister) QueryEvents 

func (p *Persister) QueryEvents(
	from, to time.Time,
	filters EventFilters,
	page, pageSize int,
	sortField string,
	sortDesc bool,
) ([]FlatDetail, int, error)

QueryEvents serves /usage/events with SQL-native pagination.

func (*Persister) RecalculateCosts 

func (p *Persister) RecalculateCosts() (RecalculateCostsResult, error)

RecalculateCosts re-prices every event using the current PriceFunc, then rebuilds hour_bucket and day_bucket from the updated events table. Heavy SQL — this is intended for occasional admin-triggered recalculation, not per-request use.

func (*Persister) Record 

func (p *Persister) Record(detail FlatDetail)

Record submits a usage record to the writer goroutine. Non-blocking — the writer drops on overflow rather than back-pressuring the request path.

func (*Persister) Retention 

func (p *Persister) Retention() config.UsageRetention

Retention returns the current retention configuration.

func (*Persister) Save 

func (p *Persister) Save()

Save flushes the WAL to the main database file. Non-blocking, no guarantee writers are quiesced — the WAL checkpoint is "PASSIVE" so it completes only as much as it can without forcing other connections to wait.

func (*Persister) SchemaVersion 

func (p *Persister) SchemaVersion(ctx context.Context) (string, error)

SchemaVersion returns the current schema_version meta value.

func (*Persister) SetPriceFunc 

func (p *Persister) SetPriceFunc(fn PriceFunc)

SetPriceFunc registers (or replaces) the pricing function.

func (*Persister) SetRetention 

func (p *Persister) SetRetention(r config.UsageRetention)

SetRetention updates retention. Takes effect on the next Trim invocation.

func (*Persister) Snapshot 

func (p *Persister) Snapshot() StatisticsSnapshot

Snapshot reconstructs a v1-style StatisticsSnapshot from the SQLite tables so /usage/export keeps returning the same response shape the UI knows how to consume. Heavy on memory for big datasets — same as the JSON-era path it replaces, but bounded by the data the user already has.

func (*Persister) SnapshotFiltered 

func (p *Persister) SnapshotFiltered(from, to time.Time, filters EventFilters) StatisticsSnapshot

SnapshotFiltered reconstructs a v1-style StatisticsSnapshot scoped by the given time range and event filters. Zero-valued from/to means "no time bound"; an empty EventFilters means "no dimension filter". The filtered version is what /usage/export uses when the user wants to export only the data they're currently viewing in the dashboard.

func (*Persister) Start 

func (p *Persister) Start(ctx context.Context)

Start opens the SQLite store, performs the legacy-JSON migration when needed, then launches the writer goroutine and the day-bucket rollup loop. Failure to open the store is logged with a clear "degraded mode" warning; the caller MUST check IsReady() afterwards and fall back to the in-memory RequestStatistics path (by calling SetGlobalPersister(nil)) when not ready, otherwise records will be silently dropped via Record's nil-writer guard.

func (*Persister) Stop 

func (p *Persister) Stop()

Stop drains the writer, halts the rollup loop, and closes the store. Idempotent.

func (*Persister) Trim 

func (p *Persister) Trim()

Trim deletes events / hour_bucket / day_bucket rows older than retention.

func (*Persister) TrimPreview 

func (p *Persister) TrimPreview() TrimPreviewResult

TrimPreview reports what Trim would delete without actually deleting it.

type PriceFunc 

type PriceFunc func(model string) (prompt, completion, cache float64, found bool)

PriceFunc returns the pricing for a model: prices are in USD per 1M tokens.

func BuildModelPriceFunc 

func BuildModelPriceFunc(configPath string) PriceFunc

BuildModelPriceFunc creates a cached PriceFunc that reads model prices from model-prices.json. When the file is missing, empty, or invalid we return nil so callers do not pretend pricing is available and accidentally emit flat-zero cost charts.

type QueryResult 

type QueryResult struct {
	Totals                 summaryTotals
	ByModel                map[string]*summaryModelStats
	ByCredential           map[string]*summaryCredentialStats
	ByAPIKey               map[string]*summaryAPIKeyStats
	TimeSeries             []timePoint
	TimeSeriesByModel      map[string][]timePoint
	TimeSeriesByCredential map[string][]timePoint
	TimeSeriesByAPIKey     map[string][]timePoint
}

QueryResult is the read-side payload Query produces and the management handler converts into JSON. It mirrors what summary.go used to expose so /usage/summary's response shape doesn't change with the SQLite swap.

type RecalculateCostsResult 

type RecalculateCostsResult struct {
	RecalculatedDays int     `json:"recalculated_days"`
	TotalCost        float64 `json:"total_cost"`
	AlreadyRunning   bool    `json:"already_running,omitempty"`
}

RecalculateCostsResult holds the outcome of a cost recalculation.

type RequestStatistics 

type RequestStatistics struct {
	// contains filtered or unexported fields
}

RequestStatistics maintains aggregated request metrics in memory. This is the legacy storage used when no Persister is configured.

func GetRequestStatistics 

func GetRequestStatistics() *RequestStatistics

GetRequestStatistics returns the shared statistics store.

func (*RequestStatistics) MergeSnapshot 

func (s *RequestStatistics) MergeSnapshot(snapshot StatisticsSnapshot, hooks ...ImportHook) MergeResult

MergeSnapshot merges an exported statistics snapshot into the current store. Existing data is preserved and duplicate request details are skipped. An optional hook is called for each newly added detail (after dedup, before return).

func (*RequestStatistics) Record 

func (s *RequestStatistics) Record(ctx context.Context, record coreusage.Record)

Record ingests a new usage record and updates the aggregates (legacy path).

func (*RequestStatistics) Snapshot 

func (s *RequestStatistics) Snapshot() StatisticsSnapshot

Snapshot returns a copy of the aggregated metrics for external consumption.

func (*RequestStatistics) TrimDetails 

func (s *RequestStatistics) TrimDetails(cutoff time.Time) []ArchivedDetail

TrimDetails removes all details older than the cutoff time from memory. Aggregate counters are preserved. Trimmed records are returned for archival.

type StatisticsSnapshot 

type StatisticsSnapshot struct {
	TotalRequests int64 `json:"total_requests"`
	SuccessCount  int64 `json:"success_count"`
	FailureCount  int64 `json:"failure_count"`
	TotalTokens   int64 `json:"total_tokens"`

	APIs map[string]aPISnapshot `json:"apis"`

	RequestsByDay  map[string]int64 `json:"requests_by_day"`
	RequestsByHour map[string]int64 `json:"requests_by_hour"`
	TokensByDay    map[string]int64 `json:"tokens_by_day"`
	TokensByHour   map[string]int64 `json:"tokens_by_hour"`
}

StatisticsSnapshot represents an immutable view of the aggregated metrics.

type Store 

type Store struct {
	// contains filtered or unexported fields
}

Store owns the *sql.DB plus dimension-name → id caches.

Cache strategy: every dim is INSERT-OR-IGNORE on first write and SELECT-d back to learn the assigned id, so id stability is guaranteed even when two goroutines race on the same name. Once seen, the (name, id) pair is pinned in the local sync.Map so steady-state writes only need a single SQL roundtrip (the events INSERT). The cache never invalidates because dim ids never change — rows are never deleted (FK ON DELETE RESTRICT) and ROWIDs are stable.

func OpenStore 

func OpenStore(dbPath string) (*Store, error)

OpenStore opens (or creates) the SQLite database, applies PRAGMAs, runs the schema (idempotent thanks to IF NOT EXISTS), and validates schema_version.

Connection limits: SQLite serialises writes regardless of pool size, so a large pool just inflates context overhead. Eight connections gives readers breathing room without wasting memory.

func (*Store) Close 

func (s *Store) Close() error

Close releases the underlying database. Safe to call on nil.

func (*Store) DB 

func (s *Store) DB() *sql.DB

DB exposes the underlying *sql.DB for writer / query / import packages.

func (*Store) Query 

func (s *Store) Query(
	ctx context.Context,
	from, to time.Time,
	filters EventFilters,
	granularity string,
	includeGroups bool,
) (QueryResult, error)

Query implements /usage/summary against the bucket tables. Routing rule: span ≤ 90d → hour_bucket; span > 90d → day_bucket. granularity controls time-series granularity but never falls back to scanning events.

func (*Store) QueryEvents 

func (s *Store) QueryEvents(
	ctx context.Context,
	from, to time.Time,
	filters EventFilters,
	page, pageSize int,
	sortField string,
	desc bool,
) ([]FlatDetail, int, error)

QueryEvents serves /usage/events. Pagination is SQL-native (LIMIT/OFFSET) so large windows don't pull every row into memory the way DetailStore.QueryRange did — that "read everything then slice" pattern was the OOM accomplice in the JSON era and explicitly cannot be reintroduced.

func (*Store) ResolveAPIKeyID 

func (s *Store) ResolveAPIKeyID(ctx context.Context, exec dbExec, name string) (sql.NullInt64, error)

ResolveAPIKeyID returns the dim_api_key id, or a NULL when name is blank.

func (*Store) ResolveAuthIndexID 

func (s *Store) ResolveAuthIndexID(ctx context.Context, exec dbExec, name string) (sql.NullInt64, error)

ResolveAuthIndexID returns the dim_auth_index id, or a NULL when name is blank.

func (*Store) ResolveCredentialID 

func (s *Store) ResolveCredentialID(ctx context.Context, exec dbExec, source, authIndex string) (sql.NullInt64, error)

ResolveCredentialID returns the dim_credential id for COALESCE(source, auth_index). Both inputs blank → NULL credential, matching summary.go::credKey semantics.

func (*Store) ResolveModelID 

func (s *Store) ResolveModelID(ctx context.Context, exec dbExec, name string) (int64, error)

ResolveModelID returns the dim_model id, falling back to "unknown" when the caller passes an empty model name. Model is the only dim that is NOT NULL on the events table, so we never return NullInt64 here.

func (*Store) ResolveProviderID 

func (s *Store) ResolveProviderID(ctx context.Context, exec dbExec, name string) (sql.NullInt64, error)

ResolveProviderID returns the dim_provider id, or a NULL when name is blank. Old v1/v2 JSON imports never carried provider, so a blank string is the expected steady-state for backfilled rows — the NULL surfaces in /usage queries as the "(unknown)" bucket so the dimension chart doesn't drop them.

func (*Store) ResolveSourceID 

func (s *Store) ResolveSourceID(ctx context.Context, exec dbExec, name string) (sql.NullInt64, error)

ResolveSourceID returns the dim_source id, or a NULL when name is blank.

func (*Store) RollupDay 

func (s *Store) RollupDay(ctx context.Context, dayStartNs int64) error

RollupDay recomputes day_bucket rows for the day starting at dayStartNs from the corresponding hour_bucket rows. Idempotent: re-running with the same day produces identical state, so missed runs after a crash are self-healing.

The UPSERT uses "SET col = excluded.col" rather than "+= excluded.col" because excluded already holds the recomputed total — adding would double count on the second pass.

func (*Store) StartRollup 

func (s *Store) StartRollup(ctx context.Context) func()

StartRollup launches the day-bucket rollup goroutine and returns a stop function. The first catch-up pass still runs synchronously before this function returns, but it now resumes from meta.day_bucket_rollup_watermark_ns instead of re-scanning every historical day on every startup. The hourly ticker keeps re-rolling yesterday/today in the background and advances the watermark as new closed days appear.

type TrimPreviewResult 

type TrimPreviewResult struct {
	FilesCount     int                `json:"files_count"`
	TotalSizeBytes int64              `json:"total_size_bytes"`
	DateRange      *trimDateRange     `json:"date_range,omitempty"`
	Details        []cleanPreviewFile `json:"details"`
}

TrimPreviewResult mirrors the JSON-era preview shape so /usage/trim/preview continues to work without UI changes.

type Writer 

type Writer struct {
	// contains filtered or unexported fields
}

Writer ingests FlatDetail records and persists them in batched transactions against a Store. A single goroutine owns all SQL writes so we never collide with SQLite's "one writer at a time" rule.

func NewWriter 

func NewWriter(store *Store, priceFn PriceFunc) *Writer

NewWriter wires a writer to a store. priceFn may be nil — events still land on disk, just with cost_micro = 0 until pricing is configured.

func (*Writer) Dropped 

func (w *Writer) Dropped() int64

Dropped exposes the lifetime drop count for diagnostics.

func (*Writer) SetPriceFunc 

func (w *Writer) SetPriceFunc(fn PriceFunc)

SetPriceFunc lets callers swap pricing at runtime (e.g. after a model catalog refresh) without restarting the writer.

func (*Writer) Start 

func (w *Writer) Start(ctx context.Context)

Start launches the writer goroutine. Calling Start more than once is a no-op so callers don't need their own guard.

func (*Writer) Stop 

func (w *Writer) Stop()

Stop closes the input channel, lets the loop drain anything in-flight, then waits for shutdown. Idempotent.

func (*Writer) Submit 

func (w *Writer) Submit(detail FlatDetail)

Submit enqueues a record for persistence. Non-blocking: when the channel is saturated the record is counted (and warning-logged with throttling) rather than back-pressuring the request path. Usage tracking is observability — a few dropped records is preferable to slowing production traffic.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.