eventcapture

package
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Sep 8, 2026 License: AGPL-3.0 Imports: 12 Imported by: 0

Documentation

Overview

Package eventcapture records high-volume operational events for offline analysis — model training data, usage matrices, replayable traces — without ever slowing the request path that produces them. It is distinct from analytics (low-volume product events to PostHog/Segment) and heavier than logging: the write path here is designed for one event per served request. The package is generic over the caller's event type; nothing about the event's shape or meaning is prescribed.

The contract that shapes everything: capture must never block or fail the hot path. Record is a non-blocking bounded-channel send — a full buffer drops the event and counts the drop rather than waiting — and a single flusher goroutine consumes the channel, writing records through a pluggable Sink (JSONL file in eventcapture/jsonl today; an object-store exporter implements the same three methods). Sink errors are logged, never surfaced: the request that produced the event has long since been answered.

Because nothing here fails loudly, the metrics are the only way to learn that capture has broken. Pass WithMetricsProvider and watch eventcapture_sink_errors (the sink is rejecting records), eventcapture_records_dropped (producers are outrunning the flusher — raise WithBufferSize or lower WithFlushInterval), and eventcapture_aggregation_overflow (a composition hit its key bound). Drops are accumulated on the hot path with an atomic and reported to the instrument at flush time, so Record itself never pays for an instrument call. Flushes are not traced — a root span every few seconds, parented to nothing, is noise — but Close is, since abandoning a drain at shutdown loses captured events.

Recorder.Run deliberately takes no context: tied to a server context it would stop consuming before the server finished draining in-flight requests, silently dropping their events. Instead the owner calls Close after the server has shut down; Close drains the buffer, runs a final flush, and closes the sink.

Aggregator folds events into per-(key, time-bucket) rollups for consumers that want densities instead of (or alongside) raw events; the key and counter types are the caller's. It is deliberately lock-free and must only be touched from the flusher goroutine — compose it through WithObserver (fold each event), WithOnFlush (emit completed buckets), and WithOverflowSource (report observations discarded at the key bound), all three of which run there. See ExampleNewRecorder_aggregation for the full composition.

Note that WithObserver names the per-event hook and has nothing to do with observability.Observer, which the Recorder builds internally from the logger and tracer provider it is given.

Example

Example captures two events and drains them. Record is a non-blocking channel send, so the request path pays nothing but the send; the flusher goroutine started by Run does the projecting and writing. 

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/primandproper/primitives-go/eventcapture"
)

// servedRequest is the caller's event type. eventcapture prescribes nothing
// about an event's shape or meaning.
type servedRequest struct {
	At     time.Time
	Route  string
	Status int
}

// sliceSink is a Sink that keeps records in memory. Deployments use
// eventcapture/jsonl or an exporter instead; the contract is these same three
// methods, all called from the Recorder's single flusher goroutine.
type sliceSink struct {
	lines []string
}

func (s *sliceSink) Write(record any) error {
	line, err := json.Marshal(record)
	if err != nil {
		return err
	}
	s.lines = append(s.lines, string(line))

	return nil
}

func (s *sliceSink) Flush() error { return nil }

func (s *sliceSink) Close() error { return nil }

// wireRequest is the projection actually written to the sink: a wire-shaped
// struct with stable JSON tags, built off the hot path.
type wireRequest struct {
	Route  string `json:"route"`
	Status int    `json:"status"`
}

func main() {
	sink := &sliceSink{}

	rec, err := eventcapture.NewRecorder[servedRequest](sink,
		eventcapture.WithTransform(func(ev *servedRequest) any {
			return wireRequest{Route: ev.Route, Status: ev.Status}
		}),
	)
	if err != nil {
		panic(err)
	}

	go rec.Run()

	rec.Record(&servedRequest{Route: "/widgets", Status: 200})
	rec.Record(&servedRequest{Route: "/widgets/1", Status: 404})

	// Close belongs after the server has finished draining in-flight requests,
	// not tied to a server context: it empties the buffer, runs a final flush,
	// and closes the sink.
	if err = rec.Close(context.Background()); err != nil {
		panic(err)
	}

	for _, line := range sink.lines {
		fmt.Println(line)
	}
	fmt.Println("dropped:", rec.Dropped())
}

Output:
{"route":"/widgets","status":200}
{"route":"/widgets/1","status":404}
dropped: 0

Index

Examples

Constants

View Source 
const (
	// DefaultBufferSize caps the in-flight event channel when WithBufferSize
	// is not supplied.
	DefaultBufferSize = 1024
	// DefaultFlushInterval is the flusher tick cadence when WithFlushInterval
	// is not supplied.
	DefaultFlushInterval = 5 * time.Second
)

Variables

View Source 
var ErrEventTypeMismatch = platformerrors.New("function event type does not match recorder type")

ErrEventTypeMismatch indicates WithTransform or WithObserver was given a function for an event type other than the Recorder's. Option carries no type parameter, so the compiler cannot catch this; NewRecorder reports it instead.

Functions

This section is empty.

Types

type Aggregator 

type Aggregator[K comparable, C any] struct {
	// contains filtered or unexported fields
}

Aggregator folds events into per-(key, time-bucket) counters of the caller's type. It is deliberately lock-free: ownership belongs to a single goroutine — the Recorder's flusher, via WithObserver/WithOnFlush — by construction. The cell map is bounded by maxKeys: once full, observations for cells not already present are dropped and counted in overflow.

func NewAggregator 

func NewAggregator[K comparable, C any](bucket time.Duration, maxKeys int, opts ...AggregatorOption) *Aggregator[K, C]

NewAggregator builds an Aggregator with the given window size and cell-map bound. A non-positive bucket defaults to one minute; a non-positive maxKeys is unbounded.

It panics if WithKeyOrder was given a comparison for a different key type; see that option for why this is a panic and not an error.

func (*Aggregator[K, C]) Flush 

func (a *Aggregator[K, C]) Flush(now time.Time, all bool) []Bucket[K, C]

Flush emits and removes completed buckets — those whose window ended at or before now — or every bucket when all is set (the drain path). Buckets are ordered by window start, then by WithKeyOrder when configured, so output is deterministic.

func (*Aggregator[K, C]) Observe 

func (a *Aggregator[K, C]) Observe(key K, at time.Time, fold func(*C))

Observe folds one observation into its (key, window) cell: fold receives the cell's counter (zero-valued on first observation) to mutate. When the cell map is full and the cell does not already exist, the observation is dropped and counted in overflow.

func (*Aggregator[K, C]) TakeOverflow 

func (a *Aggregator[K, C]) TakeOverflow() uint64

TakeOverflow returns and resets the count of observations dropped because the cell map was full, for periodic logging.

type AggregatorOption 

type AggregatorOption func(*aggregatorOptions)

AggregatorOption configures an Aggregator.

As with Option, it carries neither of the Aggregator's type parameters. The counter type C never appears in an option's arguments, so it could not be inferred — every call site would have to write both out by hand, WithKeyOrder[string, counts](cmp), forever.

func WithKeyOrder 

func WithKeyOrder[K comparable](cmp func(a, b K) int) AggregatorOption

WithKeyOrder supplies a comparison over keys (slices.SortFunc semantics) so Flush output is fully deterministic. Without it, buckets are ordered by window start only, with same-window order unspecified.

K is inferred from cmp, so this needs no type arguments: 

eventcapture.WithKeyOrder(strings.Compare)

It must match the Aggregator it configures. NewAggregator returns no error — it cannot fail for any other reason — so a comparison for the wrong key type panics there rather than being silently dropped, which would leave Flush quietly non-deterministic.

type Bucket 

type Bucket[K comparable, C any] struct {
	// Start is the bucket window's start, floored to the bucket size, in UTC.
	Start time.Time
	// Counts is the folded counter value for the (key, window) cell.
	Counts C
	// Key is the caller's aggregation key.
	Key K
	// Size is the bucket window size.
	Size time.Duration
}

Bucket is one flushed aggregation cell: the caller's counter value for one key in one time window.

type Option 

type Option func(*options)

Option configures a Recorder.

It carries no type parameter even though the Recorder does. Go cannot infer a type argument from a call's result type, so an Option[E] would force every call site to spell the event type out by hand — WithBufferSize[MyEvent](256) — forever. WithTransform and WithObserver are the two options that depend on the event type; they stay generic but still need no annotation, because E is inferable from the function each is handed.

func WithBufferSize 

func WithBufferSize(n int) Option

WithBufferSize caps the in-flight event channel. A full buffer drops (and counts) new events rather than ever blocking a caller.

func WithClock 

func WithClock(c clock.Clock) Option

WithClock swaps the clock driving the flush ticker. Tests generally do not need it: under testing/synctest the default clock already runs on bubble time.

func WithFlushInterval 

func WithFlushInterval(d time.Duration) Option

WithFlushInterval sets the cadence of the flusher tick.

func WithLogger 

func WithLogger(logger logging.Logger) Option

WithLogger attaches a logger for sink errors and drop reporting. It is named after the package, so capture lines are attributable in aggregate logs.

func WithMetricsProvider 

func WithMetricsProvider(metricsProvider metrics.Provider) Option

WithMetricsProvider attaches a metrics provider, enabling the eventcapture_* instruments. These are the only signal that a capture pipeline has broken: per the package contract sink errors are never returned to a caller, and dropped events never reach the sink at all.

func WithObserver 

func WithObserver[E any](fn func(*E)) Option

WithObserver runs fn for every consumed event, in the flusher goroutine. This is the composition point for an Aggregator's Observe.

E is inferred from the function, so this needs no type argument. It must match the Recorder it configures; NewRecorder returns ErrEventTypeMismatch otherwise, since Option cannot carry E for the compiler to check.

func WithOnFlush 

func WithOnFlush(fn func(now time.Time, final bool, emit func(record any))) Option

WithOnFlush runs fn on every flush tick and once more during the final drain (with final set). It runs in the flusher goroutine; emit writes a record through the sink with the Recorder's error handling. This is the composition point for emitting an Aggregator's completed buckets.

func WithOverflowSource 

func WithOverflowSource(fn func() uint64) Option

WithOverflowSource registers a function the flusher polls each tick to report observations an aggregation dropped for exceeding its key bound — pass an Aggregator's TakeOverflow. Without it, a full Aggregator discards observations silently, since the Recorder cannot see inside a composition whose key and counter types belong to the caller.

func WithTracerProvider 

func WithTracerProvider(tracerProvider tracing.Provider) Option

WithTracerProvider attaches a tracer provider. The flusher deliberately does not open a span per flush tick — a root span every few seconds, with no caller to parent it to, is noise rather than signal. The tracer is used for Close, where the drain is a real, once-per-process operation a shutdown trace wants to account for.

func WithTransform 

func WithTransform[E any](fn func(*E) any) Option

WithTransform projects each event into the record written to the sink — typically a wire-shaped struct with stable JSON tags — instead of the raw *E. It runs in the flusher goroutine, off the hot path.

E is inferred from the function, so this needs no type argument. It must match the Recorder it configures; NewRecorder returns ErrEventTypeMismatch otherwise, since Option cannot carry E for the compiler to check.

func WithoutRawRecords 

func WithoutRawRecords() Option

WithoutRawRecords disables the per-event sink write, for compositions that only emit derived records (e.g. aggregate rollups via WithOnFlush).

type Recorder 

type Recorder[E any] struct {
	// contains filtered or unexported fields
}

Recorder is the bridge between a hot path and a Sink: Record is a non-blocking bounded-channel send (a full buffer drops the event and counts it — capture never slows a request), and a single flusher goroutine (Run) consumes the channel, writing raw events and running the configured hooks. See the package documentation for the lifecycle rationale.

func NewRecorder 

func NewRecorder[E any](sink Sink, opts ...Option) (*Recorder[E], error)

NewRecorder builds a Recorder over sink. Start it with `go r.Run()` and stop it with Close. It returns an error only if the metrics provider cannot build the Recorder's instruments.

Example (Aggregation)

ExampleNewRecorder_aggregation composes an Aggregator into the flusher. WithObserver folds every event into its cell and WithOnFlush emits completed buckets — both run in the flusher goroutine, which is what makes the lock-free Aggregator safe. WithoutRawRecords keeps the individual events out of the sink, so only the rollups are written. 

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/primandproper/primitives-go/eventcapture"
)

// servedRequest is the caller's event type. eventcapture prescribes nothing
// about an event's shape or meaning.
type servedRequest struct {
	At     time.Time
	Route  string
	Status int
}

// sliceSink is a Sink that keeps records in memory. Deployments use
// eventcapture/jsonl or an exporter instead; the contract is these same three
// methods, all called from the Recorder's single flusher goroutine.
type sliceSink struct {
	lines []string
}

func (s *sliceSink) Write(record any) error {
	line, err := json.Marshal(record)
	if err != nil {
		return err
	}
	s.lines = append(s.lines, string(line))

	return nil
}

func (s *sliceSink) Flush() error { return nil }

func (s *sliceSink) Close() error { return nil }

// counts is the caller's counter type, folded per (route, minute) cell.
type counts struct {
	Requests int `json:"requests"`
	Errors   int `json:"errors"`
}

// rollup is the record emitted for one completed aggregation bucket.
type rollup struct {
	Minute string `json:"minute"`
	Route  string `json:"route"`
	Counts counts `json:"counts"`
}

func main() {
	sink := &sliceSink{}
	start := time.Date(2026, time.January, 1, 12, 0, 0, 0, time.UTC)

	agg := eventcapture.NewAggregator[string, counts](time.Minute, 10_000,
		eventcapture.WithKeyOrder(strings.Compare),
	)

	rec, err := eventcapture.NewRecorder[servedRequest](sink,
		// A flush interval longer than the example keeps the periodic tick from
		// splitting the rollups across two flushes; a real deployment leaves the
		// default cadence alone.
		eventcapture.WithFlushInterval(time.Hour),
		eventcapture.WithoutRawRecords(),
		// Without this the Aggregator silently discards observations once
		// maxKeys is reached; the Recorder cannot see inside the composition
		// on its own.
		eventcapture.WithOverflowSource(agg.TakeOverflow),
		eventcapture.WithObserver(func(ev *servedRequest) {
			agg.Observe(ev.Route, ev.At, func(c *counts) {
				c.Requests++
				if ev.Status >= 400 {
					c.Errors++
				}
			})
		}),
		eventcapture.WithOnFlush(func(now time.Time, final bool, emit func(record any)) {
			for _, b := range agg.Flush(now, final) {
				emit(rollup{
					Minute: b.Start.Format(time.RFC3339),
					Route:  b.Key,
					Counts: b.Counts,
				})
			}
		}),
	)
	if err != nil {
		panic(err)
	}

	go rec.Run()

	rec.Record(&servedRequest{At: start.Add(10 * time.Second), Route: "/widgets", Status: 200})
	rec.Record(&servedRequest{At: start.Add(20 * time.Second), Route: "/widgets", Status: 500})
	rec.Record(&servedRequest{At: start.Add(30 * time.Second), Route: "/gadgets", Status: 200})

	// The final flush passes all=true, so the still-open minute is emitted
	// rather than lost at shutdown.
	if err = rec.Close(context.Background()); err != nil {
		panic(err)
	}

	for _, line := range sink.lines {
		fmt.Println(line)
	}
}

Output:
{"minute":"2026-01-01T12:00:00Z","route":"/gadgets","counts":{"requests":1,"errors":0}}
{"minute":"2026-01-01T12:00:00Z","route":"/widgets","counts":{"requests":2,"errors":1}}

func (*Recorder[E]) Close 

func (r *Recorder[E]) Close(ctx context.Context) error

Close stops the flusher and waits for it to drain buffered events and close the sink, up to ctx's deadline. Safe to call more than once.

This is the one traced operation in the package: the drain is a real, once-per-process step that a shutdown trace wants accounted for, and a deadline hit here means captured events were abandoned.

func (*Recorder[E]) Dropped 

func (r *Recorder[E]) Dropped() uint64

Dropped reports how many events have been dropped because the buffer was full.

func (*Recorder[E]) Record 

func (r *Recorder[E]) Record(ev *E)

Record hands one event to the flusher. It never blocks: when the buffer is full the event is dropped and counted instead. The event is copied into the buffer; the pointer is not retained.

func (*Recorder[E]) Run 

func (r *Recorder[E]) Run()

Run is the flusher loop: it consumes events, ticks the periodic flush, and on Close drains the buffer, flushes everything, and closes the sink. Run returns only after Close is called.

type Sink 

type Sink interface {
	Write(record any) error
	// Flush pushes buffered records toward durable storage; the Recorder
	// calls it on every tick so a tail -f of a file sink stays current.
	Flush() error
	Close() error
}

Sink persists captured records. Calls arrive from the Recorder's single flusher goroutine, so implementations need no locking for Write and Flush, though Close may race a final flush and should guard itself. Write receives whatever record types the composition emits — raw events, aggregate rollups — and must not retain the value past the call.

Source Files

View all Source files

Directories

Path Synopsis
Package jsonl implements eventcapture.Sink as an append-only, size-rotated, newline-delimited JSON file.
 Package jsonl implements eventcapture.Sink as an append-only, size-rotated, newline-delimited JSON file.
Package eventcapturemock provides moq-generated mock implementations of the eventcapture package's interfaces.
 Package eventcapturemock provides moq-generated mock implementations of the eventcapture package's interfaces.
Package noop provides a no-op eventcapture.Sink, for deployments with capture wired but disabled.
 Package noop provides a no-op eventcapture.Sink, for deployments with capture wired but disabled.

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.