usage

package
v0.9.0 Latest Latest
Warning

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

Go to latest
Published: Jul 3, 2026 License: Apache-2.0 Imports: 6 Imported by: 0

Documentation

Index

Constants

View Source
const DefaultEventLimit = 100

DefaultEventLimit caps an unbounded Events query so a misconfigured caller doesn't try to deserialize the whole log file at once.

View Source
const MaxBuckets = 5_000

MaxBuckets caps the number of time buckets a single TimeSeries query may span (range / interval). Guards against a tiny interval over a huge range producing an unbounded result. The HTTP layer rejects with 400 before hitting a backend.

View Source
const MaxEventLimit = 10_000

MaxEventLimit clamps Limit to avoid OOM on a hostile request.

View Source
const MaxTagKeyLen = 64

MaxTagKeyLen caps a single tag key. Enforced at write time (app/usagelog tag validation) and reused by TagGroupKey so an unwritable key can't become a group dimension.

Variables

View Source
var ValidGroupBy = []string{
	"relay_key_hash",
	"policy_id",
	"model_id",
	"host_id",
	"host_key_id",
	"source",
	"finish_reason",
	"error_kind",
	"model",
	"host",
	"policy",
	"provider",
}

ValidGroupBy lists the accepted static GroupBy values. Used by the HTTP endpoint to reject typos with a clear 400 instead of silently grouping on nothing. "tags.<key>" is additionally accepted as a dynamic dimension — see TagGroupKey.

Functions

func IsValidGroupBy

func IsValidGroupBy(g string) bool

IsValidGroupBy reports whether g is one of the supported group dimensions (a ValidGroupBy entry or a "tags.<key>" dynamic one).

func SortTimeSeriesRows

func SortTimeSeriesRows(rows []TimeSeriesRow, totals map[string]int64, groupBy string)

SortTimeSeriesRows orders series by total request count desc, with the group value as a stable tiebreak. totals is keyed by the series key (the groupBy dimension value, or "" for the single-series case). Shared by the in-memory Bucketize path and the SQL backends so all readers return series in the same order.

func TagGroupKey added in v0.3.0

func TagGroupKey(g string) (string, bool)

TagGroupKey extracts the tag key from a dynamic "tags.<key>" group dimension. ok is false when g isn't tag-shaped or the key is empty / over MaxTagKeyLen. Backends MUST bind the returned key as a query parameter, never splice it into SQL text.

Types

type Closer

type Closer interface {
	Close() error
}

Closer is the optional shutdown contract for a Sink. The Emitter type-asserts each sink for it on Close and, when present, calls Close after the queue drains — giving buffered/remote backends (ClickHouse, valkey) a chance to flush their final batch before exit.

type DurationStats

type DurationStats struct {
	Avg int64 `json:"avg"`
	P50 int64 `json:"p50"`
	P95 int64 `json:"p95"`
	P99 int64 `json:"p99"`
	Max int64 `json:"max"`
}

DurationStats holds latency aggregates in milliseconds.

type Event

type Event struct {
	// Identity / trace
	RequestID string    `json:"request_id"`
	Source    string    `json:"source"` // "pipeline" | "proxy" | "ws" | "batch"
	Timestamp time.Time `json:"ts"`

	// Outcome
	Status       int    `json:"status"`
	DurationMs   int64  `json:"duration_ms"` // total: start → response closed
	Streamed     bool   `json:"streamed,omitempty"`
	FinishReason string `json:"finish_reason,omitempty"` // "stop"|"length"|"tool_calls"|"content_filter"|"refusal"
	Attempts     int    `json:"attempts,omitempty"`      // upstream tries (pipeline failover); 0 = not tracked
	ErrorKind    string `json:"error_kind,omitempty"`
	ErrorMessage string `json:"error_message,omitempty"`

	// Upstream is the upstream-leg timing breakdown. Nil when the request
	// never reached upstream (routing/pre-flight failure). The total
	// (start → close) is DurationMs; this adds the finer marks DurationMs
	// can't express. See UpstreamTiming for unit + how to derive TTFT.
	Upstream *sdkusage.UpstreamTiming `json:"upstream,omitempty"`

	// Reasoning is the reasoning span (when the model emitted reasoning
	// content). Nil unless the response was a canonical-observed stream
	// that carried reasoning. Microseconds from start, like Upstream.
	Reasoning *sdkusage.ReasoningTiming `json:"reasoning,omitempty"`

	// Attribution — UUIDs (stable, snapshot-resolvable to slugs at
	// query time). Hash of the inbound bearer is included so the
	// plaintext is never logged.
	RelayKeyHash   string `json:"relay_key_hash,omitempty"`
	PolicyID       string `json:"policy_id,omitempty"`
	ModelID        string `json:"model_id,omitempty"`
	RequestedModel string `json:"requested_model,omitempty"` // model string as the caller sent it
	HostID         string `json:"host_id,omitempty"`
	HostKeyID      string `json:"host_key_id,omitempty"`

	// Token usage as reported by the upstream. Empty on error or when
	// the adapter could not extract.
	Tokens sdkusage.Tokens `json:"tokens,omitempty"`

	// Free-form per-runner tags (client_ip for anonymous proxy, etc.)
	Extras map[string]string `json:"extras,omitempty"`

	// Tags are caller-owned (X-WR-Request-Tags, validated post-flight);
	// Extras stays relay-stamped — the provenance split queries rely on.
	Tags map[string]string `json:"tags,omitempty"`

	// Model, Host, Policy are the entities' slugs (metadata.name) at
	// event time, denormalized so events stay self-describing when the
	// catalog entity is gone (deleted, renamed, re-seeded). The *_id
	// fields above remain the precise keys.
	Model  string `json:"model,omitempty"`
	Host   string `json:"host,omitempty"`
	Policy string `json:"policy,omitempty"`

	// Provider is the model's owning provider slug at event time — same
	// denormalization rationale as Model/Host/Policy above.
	Provider string `json:"provider,omitempty"`

	// Cost, stamped once at emit time from the pricing then in effect.
	// CostNanos is integer nano-USD (1 USD = 1e9) — integer so sums never
	// drift; USD-only by decision, so no currency field. nil = unpriced
	// (no pricing resolved, or no priceable tokens — error/LogOnly rows),
	// which MUST stay distinguishable from a true $0: never collapse nil
	// to 0. CostBreakdown is per-meter nanos (keys like "tokens.input");
	// Pricing is the rate sheet's slug, stamped whenever one resolved
	// (even if the row ended up unpriced) for audit.
	CostNanos     *int64           `json:"cost_nanos,omitempty"`
	CostBreakdown map[string]int64 `json:"cost_breakdown,omitempty"`
	Pricing       string           `json:"pricing,omitempty"`
}

Event is the canonical per-request usage record: attribution, token counts, and the cost stamped once at emit time. Cost is a historical fact computed from the pricing in effect when the request ran — never recomputed at read time (pricing is mutable config). This package only carries the stamped fields; pricing resolution and the cost math live app-side (app/usagelog + app/pricing).

It lives in pkg/usage (not app/) so every sink/reader backend (file, ClickHouse, valkey, postgres) can consume it as a vendorable type without importing app/ — preserving the pkg-purity rule. The app/usagelog package re-exports it (type alias) for its lifecycle observers.

Field order is fixed (Go marshals in struct order) so JSONL / jq / ClickHouse column mapping stays stable across releases.

func FilterEvents

func FilterEvents(events []Event, q EventQuery) []Event

FilterEvents returns the subset of events matching q's time cutoff, dimension filters, and status range. It does not sort or apply Limit; use SortAndLimit for that.

func SortAndLimit

func SortAndLimit(events []Event, limit int) []Event

SortAndLimit sorts events newest-first and caps to limit (clamped to [DefaultEventLimit, MaxEventLimit] when out of range).

func (Event) LogOnly added in v0.3.0

func (e Event) LogOnly() bool

LogOnly reports whether the event records a request rejected before any upstream was reached: status 0 with an ErrorKind set (routing denials, auth/proxy gating, key-pool exhaustion). Such events stay visible in event listings (the logs view) but are excluded from Summary and TimeSeries aggregation — they carry no usage and would otherwise pollute stats as phantom unattributed requests.

type EventQuery

type EventQuery struct {
	// Time bounds. Lower bound resolution: From if set, else now()-Since,
	// else unbounded. Upper bound: To if set, else unbounded. From/To are
	// absolute and take precedence over the relative Since.
	Since time.Duration
	From  time.Time
	To    time.Time

	// RequestID is an exact single-value lookup (deep-link one event).
	RequestID string

	// Categorical filters — match any value in the slice (SQL IN / set
	// membership). Empty slice means no filter on that dimension.
	RelayKeyHash []string
	PolicyID     []string
	ModelID      []string
	HostID       []string
	Source       []string // "pipeline" | "proxy" | "ws" | "batch"
	FinishReason []string
	ErrorKind    []string

	// StatusMin / StatusMax restrict by HTTP status. Zero values mean
	// unbounded on that side. StatusMin=400 picks errors only;
	// StatusMin=200, StatusMax=299 picks successes only.
	StatusMin int
	StatusMax int
	// Status matches exact HTTP status codes (IN). Empty = no filter.
	// Composes with StatusMin/Max (AND).
	Status []int

	// More categorical filters (set membership; empty = no filter).
	HostKeyID      []string
	RequestedModel []string

	// Slug filters — match the denormalized entity names (Event.Model /
	// Host / Policy / Provider) recorded at event time.
	Model    []string
	Host     []string
	Policy   []string
	Provider []string

	// Tags filters on caller-supplied event tags: key → accepted values.
	// AND across keys, OR within a key's values. An event with the key
	// missing matches only an explicit "" value. Nil/empty = no filter.
	Tags map[string][]string

	// Streamed / ErrorsOnly are tri-state: nil = no filter, else match the
	// bool. ErrorsOnly true matches status>=400 OR error_kind!="" (false
	// matches the complement — "successes only").
	Streamed   *bool
	ErrorsOnly *bool

	// AttemptsMin filters by upstream try count (failover). 0 = no filter.
	AttemptsMin int

	// DurationMsMin / DurationMsMax bound total request duration (ms).
	DurationMsMin int64
	DurationMsMax int64

	// TTFTMsMin / TTFTMsMax bound upstream time-to-first-byte (ms), derived
	// from Upstream.ResponseStart (µs from start). Events with no upstream
	// timing are excluded when either bound is set.
	TTFTMsMin int64
	TTFTMsMax int64

	// Q is a free-text needle matched (case-insensitive substring) against
	// request_id, model_id, requested_model, and source.
	Q string

	// Limit caps the number of events returned. <=0 → DefaultEventLimit.
	Limit int

	// CursorTS / CursorID implement keyset pagination for Events. When
	// CursorTS is non-zero, only events strictly older than the cursor are
	// returned — i.e. (ts, request_id) < (CursorTS, CursorID) under the
	// (ts DESC, request_id DESC) ordering. Set only on the Events path;
	// ignored by Summary / TimeSeries.
	CursorTS time.Time
	CursorID string
}

EventQuery filters the raw event stream. All fields are optional; zero values mean "no filter on this dimension."

type Reader

type Reader interface {
	// Events returns raw events matching q, in reverse-chronological
	// order (newest first), capped at q.Limit.
	Events(ctx context.Context, q EventQuery) ([]Event, error)

	// Summary returns aggregated rows grouped by q.GroupBy. Each row
	// carries totals + latency percentiles over the events matching
	// the filter. Rows are sorted by Requests descending. LogOnly
	// events (pre-upstream rejections) are excluded from aggregation —
	// every backend must apply the Event.LogOnly predicate. The same
	// exclusion applies to TimeSeries; Events listings keep them.
	Summary(ctx context.Context, q SummaryQuery) (SummaryResult, error)

	// TimeSeries returns one or more series of time-bucketed aggregates
	// over the filtered set. With an empty GroupBy a single series is
	// returned; with a GroupBy dimension, one series per distinct value.
	// Buckets are aligned to the Unix epoch by q.Interval and ordered
	// oldest-first within each series.
	TimeSeries(ctx context.Context, q TimeSeriesQuery) (TimeSeriesResult, error)
}

Reader is the read-side counterpart to Sink. Implementations satisfy this against whatever backing store carries usage events — JSONL file, ClickHouse, valkey, in-memory for tests. Consumers (HTTP endpoints, CLI) depend on the interface, not the backend.

type Sink

type Sink interface {
	// Write delivers one event. The error is logged by the Emitter;
	// returning non-nil does not stop subsequent events.
	Write(ev Event) error
}

Sink consumes usage events. Implementations are expected to be non-blocking from the Emitter drain goroutine's perspective — slow I/O belongs inside the sink's own buffering, not in the caller.

A backend that needs a graceful drain (flush buffered rows, close a remote connection) additionally implements Closer; the Emitter calls Close after draining at shutdown.

type SummaryQuery

type SummaryQuery struct {
	EventQuery

	// GroupBy is the dimension to group on. Valid values:
	// "relay_key_hash", "policy_id", "model_id", "host_id",
	// "host_key_id", "source", "finish_reason", "error_kind",
	// "model", "host", "policy", "provider" (event-time slugs),
	// or "tags.<key>" (dynamic, groups on a caller tag's value).
	// Empty → "source".
	GroupBy string
}

SummaryQuery is EventQuery + group dimension. The filter fields are embedded so reader implementations can share filter logic.

type SummaryResult

type SummaryResult struct {
	Rows []SummaryRow `json:"rows"`
	From time.Time    `json:"from"`
	To   time.Time    `json:"to"`
}

SummaryResult wraps the rows with the resolved time range so the caller can render "events from X to Y" without re-deriving from the query.

func Summarize

func Summarize(events []Event, groupBy string) (SummaryResult, error)

Summarize groups the (already-filtered) events by groupBy and builds per-group totals + latency percentiles, sorted by request count desc. LogOnly events (pre-upstream rejections) are skipped — they belong to the logs view, not usage stats. Returns an error for an unknown groupBy dimension.

type SummaryRow

type SummaryRow struct {
	Group      map[string]string `json:"group"`
	Requests   int64             `json:"requests"`
	ErrorCount int64             `json:"error_count"`
	Tokens     map[string]int64  `json:"tokens"`
	DurationMs DurationStats     `json:"duration_ms"`
	// TTFTMs aggregates upstream time-to-first-byte (ms, derived from
	// Upstream.ResponseStart) over the subset of events that carry upstream
	// timing. Nil when no event in the group has it — a pointer so "no
	// samples" is distinguishable from "all-zero latency".
	TTFTMs    *DurationStats `json:"ttft_ms,omitempty"`
	FirstSeen time.Time      `json:"first_seen"`
	LastSeen  time.Time      `json:"last_seen"`
	// CostNanos sums Event.CostNanos over the group's priced events
	// (nano-USD). Unpriced counts the events that carried no cost stamp —
	// reported instead of folding them into the sum as silent zeros, so a
	// cost total always says how complete it is.
	CostNanos int64 `json:"cost_nanos"`
	Unpriced  int64 `json:"unpriced"`
}

SummaryRow is one grouped aggregate. The Group map carries the grouping dimension(s) keyed by their column name — single-key today, extensible to multi-key grouping later without changing the type.

type TimeSeriesPoint

type TimeSeriesPoint struct {
	Bucket     time.Time `json:"bucket"`
	Requests   int64     `json:"requests"`
	ErrorCount int64     `json:"error_count"`
	// Errors4xx/Errors5xx split ErrorCount by status class for triage
	// charts. They may sum below ErrorCount only if an out-of-band status
	// >= 600 ever appears; in practice ErrorCount = Errors4xx + Errors5xx.
	Errors4xx  int64            `json:"errors_4xx"`
	Errors5xx  int64            `json:"errors_5xx"`
	Tokens     map[string]int64 `json:"tokens"`
	DurationMs DurationStats    `json:"duration_ms"`
	// TTFTMs — see SummaryRow.TTFTMs; nil when no event in the bucket
	// carries upstream timing.
	TTFTMs *DurationStats `json:"ttft_ms,omitempty"`
	// CostNanos / Unpriced — see SummaryRow.
	CostNanos int64 `json:"cost_nanos"`
	Unpriced  int64 `json:"unpriced"`
}

TimeSeriesPoint is one time bucket's aggregates. Bucket is the bucket's start instant (UTC, epoch-aligned). Empty buckets are omitted — the frontend zero-fills gaps against the resolved From/To range.

type TimeSeriesQuery

type TimeSeriesQuery struct {
	EventQuery

	// Interval is the bucket width. Must be > 0.
	Interval time.Duration

	// GroupBy splits the result into one series per dimension value.
	// Empty means a single series. Valid values match ValidGroupBy.
	GroupBy string
}

TimeSeriesQuery is EventQuery + a bucket width + optional group dimension. Interval is required (zero is rejected by readers). GroupBy is optional: empty yields a single series, a valid dimension yields one series per distinct value.

type TimeSeriesResult

type TimeSeriesResult struct {
	Rows     []TimeSeriesRow `json:"rows"`
	Interval string          `json:"interval"`
	From     time.Time       `json:"from"`
	To       time.Time       `json:"to"`
}

TimeSeriesResult wraps the series with the resolved interval (echoed as a string, e.g. "1h") and time range so the caller can zero-fill and label the axis without re-deriving from the query.

func Bucketize

func Bucketize(events []Event, interval time.Duration, groupBy string) (TimeSeriesResult, error)

Bucketize groups the (already-filtered) events into time buckets of width interval, optionally split into one series per groupBy dimension. Buckets align to the Unix epoch (bucketStart = floor(unix/intervalSec)); empty buckets are omitted. Points within a series are ordered oldest-first; series (rows) are ordered by total request count desc. Returns an error for a non-positive interval or unknown groupBy.

type TimeSeriesRow

type TimeSeriesRow struct {
	Group  map[string]string `json:"group,omitempty"`
	Points []TimeSeriesPoint `json:"points"`
}

TimeSeriesRow is one series. Group is nil for the single-series case and carries the grouping dimension keyed by column name otherwise.

Directories

Path Synopsis
Package clickhouse provides a ClickHouse-backed implementation of usage.Sink, usage.Reader, and usage.Closer.
Package clickhouse provides a ClickHouse-backed implementation of usage.Sink, usage.Reader, and usage.Closer.
Package file is the JSONL file backend for usage logging: a Sink that appends one event per line and a Reader that scans the same file.
Package file is the JSONL file backend for usage logging: a Sink that appends one event per line and a Reader that scans the same file.
Package postgres implements usage.Sink, usage.Reader, and usage.Closer backed by PostgreSQL.
Package postgres implements usage.Sink, usage.Reader, and usage.Closer backed by PostgreSQL.
Package valkey is a TTL-bounded usage Sink + Reader backed by pkg/kv.
Package valkey is a TTL-bounded usage Sink + Reader backed by pkg/kv.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL