usage

package
v0.1.0 Latest Latest
Warning

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

Go to latest
Published: Jun 3, 2026 License: AGPL-3.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.

Variables

View Source
var ValidGroupBy = []string{
	"relay_key_hash",
	"policy_id",
	"model_id",
	"host_id",
	"host_key_id",
	"source",
}

ValidGroupBy lists the accepted GroupBy values. Used by the HTTP endpoint to reject typos with a clear 400 instead of silently grouping on nothing.

Functions

func IsValidGroupBy

func IsValidGroupBy(g string) bool

IsValidGroupBy reports whether g is one of the supported group dimensions.

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.

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"`
}

Event is the canonical per-request usage record. Pure attribution + token counts — no cost, no pricing, no derived business metrics. Cost computation, billing, and analytics are downstream consumer concerns; they read this event and apply their own logic.

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).

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

	// 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.
	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". 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. 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"`
	FirstSeen  time.Time         `json:"first_seen"`
	LastSeen   time.Time         `json:"last_seen"`
}

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"`
	Tokens     map[string]int64 `json:"tokens"`
}

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