Documentation
¶
Overview ¶
Package usagelog is the first consumer of pkg/lifecycle's Hook + Collector surface. Per-request usage records — pure attribution + token counts.
Four pieces, split along produce → attach → store:
- Event: the canonical usage schema. Stable JSON shape — sinks (file JSONL today, ClickHouse later) consume by deserializing this struct directly. No cost, no pricing — those are downstream consumer concerns. The event answers "who called what model on which host and how many tokens." Cost layers on at query time.
- UsageHook: the lifecycle.Hook (producer). Parses tokens + finish reason out of the response body via v1.ExtractSummary, builds the Event, returns it. Pure — the Registry attaches it to the Context under Namespace; the hook never touches the Context.
- SinkCollector: the lifecycle.Collector (janitor / store). Reads the attached Event back off the Context and pushes it onto the Emitter.
- Emitter: bounded-queue dispatcher. One drain goroutine writes to each Sink. Drop-on-full with counter — never blocks the post-flight goroutine.
The split is what lets a pre-send reader (usage echo) and the sink share one collection: the hook produces once, the Registry attaches, and any number of readers/collectors consume it.
Index ¶
- Constants
- Variables
- func EchoFromContext(lc *lifecycle.Context) *v1.RelayUsage
- func IsValidGroupBy(g string) bool
- func ParseTags(raw string) (map[string]string, bool)
- type Backend
- type BackendBuilder
- type Closer
- type Controller
- type DurationStats
- type Emitter
- type EmitterOptions
- type Event
- type EventQuery
- type Pricer
- type PricingLookup
- type Reader
- type ReasoningTiming
- type SettingsSource
- type Sink
- type SinkCollector
- type StreamUsageFactory
- type SummaryQuery
- type SummaryResult
- type SummaryRow
- type TimeSeriesPoint
- type TimeSeriesQuery
- type TimeSeriesResult
- type TimeSeriesRow
- type UpstreamTiming
- type UsageHook
Constants ¶
const ( DefaultEventLimit = usage.DefaultEventLimit MaxEventLimit = usage.MaxEventLimit MaxBuckets = usage.MaxBuckets )
const DefaultQueueSize = 1024
DefaultQueueSize is the default bounded-channel capacity if EmitterOptions.QueueSize is zero. 1024 events comfortably absorbs a brief drain stall (sink slow / disk flush / pipe back-pressure) without blocking the post-flight goroutine.
const MaxTagsHeaderBytes = 2048
MaxTagsHeaderBytes caps the raw header at the inference edge — longer values are never copied into Metadata.
const MetadataKeyRequestTags = "request_tags_raw"
MetadataKeyRequestTags is the lifecycle Metadata key under which the inference edge stashes the raw X-WR-Request-Tags header value. The hot path copies the string only; ParseTags runs post-flight.
const Namespace = "usage"
Namespace is the key under which UsageHook attaches its Event to the lifecycle Context. SinkCollector (store side) and any pre-send reader (usage echo) read it back via lc.Collected(Namespace).
Variables ¶
var ValidGroupBy = usage.ValidGroupBy
ValidGroupBy lists the accepted GroupBy values (see pkg/usage).
Functions ¶
func EchoFromContext ¶
func EchoFromContext(lc *lifecycle.Context) *v1.RelayUsage
EchoFromContext maps the usage Event that UsageHook attached to lc into the caller-facing v1.RelayUsage block. Returns nil when no usage Event was collected (so the response-writer simply doesn't inject anything).
This is the read side of the blackboard: the producer (UsageHook) filled the Event once; echo reads that same collection here, the sink reads it at Finalize — no re-parse. Deliberately drops operator-only attribution (relay-key hash, policy/host/model UUIDs); the caller gets observability, not internals.
func IsValidGroupBy ¶
IsValidGroupBy reports whether g is a supported group dimension.
Types ¶
type Backend ¶
Backend bundles the sink (write) and reader (read) a log backend provides. For clickhouse/postgres/valkey, Sink and Reader are the same instance; for file they differ. The composition root names the backend packages and builds this; everything else consumes it through the Controller.
type BackendBuilder ¶
BackendBuilder constructs the Backend for a resolved config. Injected by the composition root so backend packages stay named only there.
type Closer ¶
Closer is the optional graceful-shutdown contract a Sink may implement; the Emitter calls it after draining. See pkg/usage.Closer.
type Controller ¶
type Controller struct {
// contains filtered or unexported fields
}
Controller owns the log backend's runtime lifecycle: a long-lived Emitter draining into a reloadable sink, plus a reloadable reader the control plane queries. It reconciles the backend (file ↔ clickhouse ↔ postgres ↔ valkey) on settings changes without a restart — a reroute is a clean break (new events go to the new store; old data stays put, no migration).
Unlike payload-logging there is no enable/maxbytes gate: logging is constant. The swap closes the old sink after in-flight writes drain (reloadableSink's RWMutex); the old reader (== old sink for ch/pg/valkey, or a resource-less file reader) needs no separate close.
func NewController ¶
func NewController(src SettingsSource, build BackendBuilder, log *slog.Logger) *Controller
NewController wires the emitter → reloadable-sink chain and the reloadable reader. The sink/reader start empty until the first reconcile.
func (*Controller) Close ¶
func (c *Controller) Close()
Close drains the emitter (which closes the live sink).
func (*Controller) Emitter ¶
func (c *Controller) Emitter() *Emitter
Emitter returns the long-lived emitter (the collector registers on it).
func (*Controller) Reader ¶
func (c *Controller) Reader() Reader
Reader returns the stable reader handle the control plane queries; it delegates to whichever backend reader is currently live.
func (*Controller) Run ¶
func (c *Controller) Run(ctx context.Context)
Run does the initial reconcile then reconciles on each kick until ctx ends.
func (*Controller) Subscribe ¶
func (c *Controller) Subscribe()
Subscribe registers the settings-change callback (cheap, non-blocking — signals the kick channel; the rebuild runs on Run's goroutine).
type DurationStats ¶
type DurationStats = usage.DurationStats
Query / result types — re-exported from pkg/usage so callers that only import usagelog keep compiling. New code may use pkg/usage directly.
type Emitter ¶
type Emitter struct {
// contains filtered or unexported fields
}
Emitter is the fan-out point: hooks Emit() events; a single background goroutine drains the queue and writes to each Sink.
Drop-on-full preserves the "post-flight never blocks" invariant. Drop counter is exposed via Dropped() for /metrics or assertions.
func NewEmitter ¶
func NewEmitter(opts EmitterOptions, sinks ...Sink) *Emitter
NewEmitter constructs an Emitter and starts its drain goroutine. The Emitter must be Closed at shutdown to flush in-flight events.
func (*Emitter) Close ¶
func (e *Emitter) Close()
Close signals the drain goroutine to finish, drains pending events, and returns once all sinks have processed everything in flight. After the queue drains, any sink implementing Closer is closed (flush final batch, close remote conn) so buffered/remote backends lose nothing on graceful shutdown. Subsequent Emit calls are no-ops.
func (*Emitter) Dropped ¶
Dropped returns the cumulative count of events dropped due to a full queue. Useful for /metrics scraping.
func (*Emitter) Emit ¶
Emit queues ev for delivery to all sinks. Non-blocking; if the queue is full the event is dropped and the drop counter increments. Safe for concurrent calls — the underlying channel handles fan-in.
func (*Emitter) QueueDepth ¶ added in v0.3.0
QueueDepth returns the number of events waiting in the bounded queue — the leading signal before Dropped starts counting. Safe to call concurrently (chan len).
type EmitterOptions ¶
type EmitterOptions struct {
// QueueSize is the bounded-channel capacity. <= 0 → DefaultQueueSize.
QueueSize int
// Logger is used for drop / sink-error warnings. nil → slog.Default.
Logger *slog.Logger
}
EmitterOptions tunes Emitter behavior. Zero values are sensible defaults for typical deployments.
type Event ¶
Event is the canonical per-request usage record. The canonical type lives in pkg/usage so every backend (file, ClickHouse, valkey, postgres) can consume it without importing app/ (pkg-purity rule). This alias keeps the lifecycle observers in this package ergonomic.
type EventQuery ¶
type EventQuery = usage.EventQuery
Query / result types — re-exported from pkg/usage so callers that only import usagelog keep compiling. New code may use pkg/usage directly.
type Pricer ¶ added in v0.3.0
type Pricer struct {
// contains filtered or unexported fields
}
Pricer maps a request's stamped pricing identity + extracted token counts to the cost fields stamped on the usage Event. Cost is computed exactly once, here at emit time, from the rate sheet currently in the catalog snapshot — what a request cost is a historical fact, so readers never recompute it against (mutable) pricing config. Nil-safe: a nil Pricer (tests, partial wiring) prices nothing, leaving events unpriced.
func NewPricer ¶ added in v0.3.0
func NewPricer(lookup PricingLookup) *Pricer
NewPricer constructs a Pricer over the given lookup.
func (*Pricer) Price ¶ added in v0.3.0
func (p *Pricer) Price(pricingID string, tokens sdkusage.Tokens) (nanos int64, breakdown map[string]int64, ok bool)
Price returns the total nano-USD cost + per-meter breakdown for the tokens under pricingID. ok=false means unpriced — no pricing id stamped, the rate sheet is gone from the snapshot, or no token key matched a rate. Unpriced is never reported as a zero cost.
type PricingLookup ¶ added in v0.3.0
PricingLookup resolves a pricing id to the rate sheet. Satisfied by a closure over app/catalog's live snapshot at the composition root — snapshot map read, zero I/O.
type Reader ¶
Reader is the read-side interface backends implement. Canonical definition lives in pkg/usage; re-exported here for the control-plane consumers (and cmd/relay-stats) that import usagelog. The concrete backends live under pkg/usage/<backend> (file today, clickhouse/valkey next).
type ReasoningTiming ¶
type ReasoningTiming = sdkusage.ReasoningTiming
ReasoningTiming is the reasoning span — see pkg/usage.ReasoningTiming.
type SettingsSource ¶
type SettingsSource interface {
Setting(section string) (any, bool)
OnSettingsChange(section string, fn func())
}
SettingsSource reads a live settings section + subscribes to changes. Satisfied by *app/catalog.Catalog.
type Sink ¶
Sink consumes usage events. Canonical interface in pkg/usage; the concrete backends (file, clickhouse, valkey, postgres) live under pkg/usage/<backend> and implement it. Re-exported here for the Emitter and the composition root.
type SinkCollector ¶
type SinkCollector struct {
// contains filtered or unexported fields
}
SinkCollector is the usage janitor: a lifecycle.Collector that reads the *Event UsageHook attached under Namespace and pushes it onto the bounded Emitter (→ Sink). Read-only on the Context; the Emit is a non-blocking channel send, so it never blocks the post-flight goroutine.
This is the "store" half of the produce → attach → store flow. Adding a second destination (ClickHouse, OTel) is a matter of the Emitter fanning to more Sinks — not a change to this collector or the hook.
func NewSinkCollector ¶
func NewSinkCollector(e *Emitter) *SinkCollector
NewSinkCollector constructs a collector that emits collected usage Events onto e.
func (*SinkCollector) Collect ¶
func (c *SinkCollector) Collect(lc *lifecycle.Context)
Collect reads the usage Event off lc and emits it. No-op when no usage Event was attached (defensive — UsageHook always attaches one).
type StreamUsageFactory ¶
type StreamUsageFactory struct {
// contains filtered or unexported fields
}
StreamUsageFactory is the streaming counterpart to UsageHook: a lifecycle.StreamObserverFactory whose observer buffers the raw upstream SSE frames and, at end-of-stream, builds the same usage Event via the shared buildEvent (which runs v1.ExtractSummary on the accumulated body). Streamed requests therefore collect usage exactly like buffered ones — same Event, same Namespace — so echo and the sink both see one shape. The post-flight Hook is too late for a streamed body (it runs after the stream is already written); this fills during the stream.
func NewStreamUsageFactory ¶
func NewStreamUsageFactory(pricer *Pricer) *StreamUsageFactory
NewStreamUsageFactory constructs the factory. pricer may be nil (events stay unpriced).
func (*StreamUsageFactory) Name ¶
func (*StreamUsageFactory) Name() string
func (*StreamUsageFactory) NewObserver ¶
func (f *StreamUsageFactory) NewObserver(lc *lifecycle.Context) lifecycle.StreamObserver
type SummaryQuery ¶
type SummaryQuery = usage.SummaryQuery
Query / result types — re-exported from pkg/usage so callers that only import usagelog keep compiling. New code may use pkg/usage directly.
type SummaryResult ¶
type SummaryResult = usage.SummaryResult
Query / result types — re-exported from pkg/usage so callers that only import usagelog keep compiling. New code may use pkg/usage directly.
type SummaryRow ¶
type SummaryRow = usage.SummaryRow
Query / result types — re-exported from pkg/usage so callers that only import usagelog keep compiling. New code may use pkg/usage directly.
type TimeSeriesPoint ¶
type TimeSeriesPoint = usage.TimeSeriesPoint
Query / result types — re-exported from pkg/usage so callers that only import usagelog keep compiling. New code may use pkg/usage directly.
type TimeSeriesQuery ¶
type TimeSeriesQuery = usage.TimeSeriesQuery
Query / result types — re-exported from pkg/usage so callers that only import usagelog keep compiling. New code may use pkg/usage directly.
type TimeSeriesResult ¶
type TimeSeriesResult = usage.TimeSeriesResult
Query / result types — re-exported from pkg/usage so callers that only import usagelog keep compiling. New code may use pkg/usage directly.
type TimeSeriesRow ¶
type TimeSeriesRow = usage.TimeSeriesRow
Query / result types — re-exported from pkg/usage so callers that only import usagelog keep compiling. New code may use pkg/usage directly.
type UpstreamTiming ¶
type UpstreamTiming = sdkusage.UpstreamTiming
UpstreamTiming is the upstream-leg timing breakdown — see pkg/usage.UpstreamTiming.
type UsageHook ¶
type UsageHook struct {
// contains filtered or unexported fields
}
UsageHook is the usage producer: a lifecycle.Hook that builds the canonical per-request Event from the Context's identity/timing and the PostFlightEvent's outcome. Pure — it never touches the Context; the Registry attaches the returned *Event under Namespace. Storing happens later, in SinkCollector.
Canonical-first: token counts + finish reason come from v1.ExtractSummary on the buffered body using the per-request Translator — no vendor-specific JSON/SSE branching at this layer.
func NewUsageHook ¶
NewUsageHook constructs the usage producer. pricer may be nil (events stay unpriced).
Source Files