requestlog

Documentation

Overview

Package requestlog records simulated-API request/response traffic to SQLite. Writes are asynchronous (off the request hot path) and bounded (ring + size cap).

Index

• func WithRing(n int) func(*Options)
• type Bus
  • func NewBus() *Bus
  • func (b *Bus) Publish(e Entry)
  • func (b *Bus) Subscribe() (<-chan Entry, func())
• type Entry
• type Options
• type Query
• type Recorder
  • func NewRecorder(st *Store, service string, seq *atomic.Int64) *Recorder
  • func (r *Recorder) Wrap(next http.Handler) http.Handler
• type Store
  • func Open(path string, opts ...func(*Options)) (*Store, error)
  • func (s *Store) Bus() *Bus
  • func (s *Store) Clear() error
  • func (s *Store) Close() error
  • func (s *Store) Enqueue(e Entry)
  • func (s *Store) Flush()
  • func (s *Store) Get(id int64) (Entry, error)
  • func (s *Store) Insert(e Entry) error
  • func (s *Store) List(q Query) ([]Entry, error)

Functions

func WithRing

func WithRing(n int) func(*Options)

WithRing sets the ring-bound (newest-N) retention.

Types

type Bus

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

Bus is a fan-out publisher of captured entries. Subscribers receive only events published AFTER they subscribe (no history); slow subscribers have events dropped (non-blocking) so a parked client never blocks recorders.

func NewBus

func NewBus() *Bus

NewBus builds an empty bus.

func (*Bus) Publish

func (b *Bus) Publish(e Entry)

Publish sends e to all subscribers, non-blocking (drops on full).

func (*Bus) Subscribe

func (b *Bus) Subscribe() (<-chan Entry, func())

Subscribe returns a channel of future entries and a cancel func to remove it. The channel is buffered (128); overflow is dropped on publish.

type Entry

type Entry struct {
	ID          int64  `json:"id"`
	Seq         int64  `json:"seq"`
	Ts          string `json:"ts"` // RFC3339
	Service     string `json:"service"`
	Transport   string `json:"transport"` // http | grpc | ws
	Method      string `json:"method"`
	Path        string `json:"path"`
	Status      int    `json:"status"`
	DurationUs  int64  `json:"duration_us"`  // microseconds (sub-ms resolution)
	ReqHeaders  string `json:"req_headers"`  // JSON, sensitive values redacted
	ReqBody     string `json:"req_body"`     // capped
	RespHeaders string `json:"resp_headers"` // JSON
	RespBody    string `json:"resp_body"`    // capped
}

Entry is one captured request/response.

type Options

type Options struct {
	Ring       int   // keep at most this many newest entries (0 = unlimited)
	MaxBytes   int64 // rotate (delete oldest) when DB exceeds this (0 = unlimited)
	WriteQueue int   // async enqueue buffer size (default 1024)
}

Options configures the store.

type Query

type Query struct {
	Service string
	Method  string
	Path    string // substring match
	Status  int    // 0 = any
	Q       string // free-text over path+req_body+resp_body
	Since   int64  // only entries with seq > Since (0 = no filter; gap-free backfill)
	Limit   int
	Offset  int
}

Query filters the history list.

type Recorder

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

Recorder wraps a handler to capture req/resp into the store.

func NewRecorder

func NewRecorder(st *Store, service string, seq *atomic.Int64) *Recorder

NewRecorder builds a recorder bound to a service name. seq is a shared, engine-wide monotonic counter so entry Seq values are globally unique (needed for the live-feed gap-free ordering in Plan 2 and unique row labels).

func (*Recorder) Wrap

func (r *Recorder) Wrap(next http.Handler) http.Handler

Wrap returns a handler that records then delegates to next.

type Store

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

Store is the SQLite-backed request log.

func Open

func Open(path string, opts ...func(*Options)) (*Store, error)

Open opens (or creates) the request log at path with the given options. Without options it defaults to Ring=1000, WriteQueue=1024.

func (*Store) Bus

func (s *Store) Bus() *Bus

Bus returns the store's publisher (recorders publish captured entries here).

func (*Store) Clear

func (s *Store) Clear() error

Clear removes every captured request (used by reset for deterministic runs). It waits for in-flight writes first so the table isn't cleared mid-write.

func (*Store) Close

func (s *Store) Close() error

Close drains the async writer (if running), stops it, then closes the DB. It is idempotent. Draining before close guarantees in-flight writes are persisted and the single connection is released, so a caller reopening the same DB (e.g. a test restarting an engine in one state dir) does not hit SQLITE_BUSY.

func (*Store) Enqueue

func (s *Store) Enqueue(e Entry)

Enqueue hands an entry to the writer goroutine (non-blocking; drops on full). If no writer is running (store opened read-only), it persists synchronously.

func (*Store) Flush

func (s *Store) Flush()

Flush blocks until all enqueued entries have been written by the writer goroutine. It is a no-op for read-only stores.

func (*Store) Get

func (s *Store) Get(id int64) (Entry, error)

Get returns a single entry by DB id (for detail + replay).

func (*Store) Insert

func (s *Store) Insert(e Entry) error

Insert persists an entry synchronously (back-compat / test path). It does NOT go through the async writer and does not touch the in-flight counter.

func (*Store) List

func (s *Store) List(q Query) ([]Entry, error)

List returns entries matching q, newest-first. Non-empty filter fields (Service, Method, Path substring, Q free-text) are AND-combined into a WHERE clause; an empty query returns the newest entries up to Limit.