Documentation
¶
Overview ¶
Package payloadlog is the second lifecycle observer (after app/usagelog): it captures the full request + response bodies of opted-in requests for audit, debug, and replay, and pushes them to a Sink (file today, S3 next) off the hot path.
It mirrors usagelog's shape exactly — a produce → attach → store flow on the shared lifecycle Registry:
- PayloadHook (buffered path): builds a Record from lc.RequestBody + PostFlightEvent.ResponseBody.
- StreamPayloadFactory (streamed path): accumulates upstream SSE frames and builds the Record at end-of-stream. Returns a no-op observer when capture is disabled, so non-opted-in streams never buffer.
- SinkCollector (janitor): reads the attached Record and emits it onto the bounded, drop-on-full Emitter → Sink.
Capture is OFF by default and gated per request via lc.PayloadLog, set at the inference entry from the routing Plan (Policy or RelayKey opt-in). Bodies are stored raw; the only transform is a configurable size cap that truncates oversized bodies and flags the Record. Operator-internal: there is no read endpoint — payloads are an offline artifact, joinable to usage events by request_id.
Index ¶
Constants ¶
const DefaultQueueSize = 256
DefaultQueueSize is the bounded-channel capacity when EmitterOptions leaves it zero. Payload Records are larger than usage events, so the default is smaller — drop-on-full still protects the post-flight goroutine; a dropped payload is acceptable (it's an audit best-effort, not billing truth).
const Namespace = "payload"
Namespace is the key under which the payload Record is attached to the lifecycle Context. SinkCollector reads it back via lc.Collected. Distinct from usagelog's "usage" so the two observers coexist on the same Registry.
Variables ¶
var ErrNotFound = payload.ErrNotFound
ErrNotFound is re-exported so control handlers can map an absent capture to 404 without importing pkg/payload directly.
Functions ¶
This section is empty.
Types ¶
type Closer ¶
Record / Sink / Closer are re-exported from pkg/payload so the observer and its wiring can import just payloadlog. The contract + backends live in pkg/payload and pkg/payload/<backend> (file, s3) — vendorable, and the s3 driver is excludable from minimal builds.
type Controller ¶
type Controller struct {
// contains filtered or unexported fields
}
Controller owns payload logging's runtime state: a single long-lived Emitter draining into a reloadable sink, plus the live enabled/maxBytes the Hook and StreamObserver read per request. It reconciles the sink (toggle / backend / bucket / credentials) on settings changes without a restart — the observer is always registered, so flipping the setting on just installs a sink. Changes are event-driven: Subscribe registers a catalog change-callback that signals Run; the rebuild (which may do network I/O) happens on Run's goroutine, never on the catalog listener.
func NewController ¶
func NewController(src SettingsSource, build SinkBuilder, log *slog.Logger) *Controller
NewController wires the emitter → reloadable-sink chain. The sink starts empty (nothing written) until the first reconcile applies the live config. reader provides the live settings; build constructs sinks.
func (*Controller) Close ¶
func (c *Controller) Close()
Close drains the emitter and closes the current sink. Idempotent via the emitter's own guard.
func (*Controller) Emitter ¶
func (c *Controller) Emitter() *Emitter
Emitter is the bounded queue the SinkCollector emits into.
func (*Controller) Enabled ¶
func (c *Controller) Enabled() bool
Enabled reports the live master switch. The Hook/StreamObserver gate on this AND the per-request opt-in (lc.PayloadLog).
func (*Controller) MaxBytes ¶
func (c *Controller) MaxBytes() int
MaxBytes is the live per-body storage cap (0 = unlimited).
func (*Controller) Run ¶
func (c *Controller) Run(ctx context.Context)
Run reconciles once, then on every signal from Subscribe's callback, until ctx is cancelled. Call in a goroutine after Subscribe. The rebuild runs here, off the catalog listener.
func (*Controller) Subscribe ¶
func (c *Controller) Subscribe()
Subscribe registers the controller for payload-logging settings changes. Call it synchronously during composition, before the catalog Hydrate runs, so the boot reload's notification isn't missed. The callback only signals Run; it never rebuilds inline, keeping slow sink I/O off the catalog's serial listener goroutine.
type Emitter ¶
type Emitter struct {
// contains filtered or unexported fields
}
Emitter fans captured Records to its Sinks via a single drain goroutine. Drop-on-full preserves the "post-flight never blocks" invariant; the drop count is exposed via Dropped().
func NewEmitter ¶
func NewEmitter(opts EmitterOptions, sinks ...Sink) *Emitter
NewEmitter constructs an Emitter and starts its drain goroutine. Close it at shutdown to flush in-flight Records.
func (*Emitter) Close ¶
func (e *Emitter) Close()
Close drains pending Records, then closes any Closer sink. Subsequent Emit calls are no-ops.
func (*Emitter) Dropped ¶
Dropped returns the cumulative count of Records dropped due to a full queue.
func (*Emitter) Emit ¶
Emit queues r for delivery. Non-blocking; a full queue drops the Record and increments the drop counter. Safe for concurrent calls.
func (*Emitter) QueueDepth ¶ added in v0.3.0
QueueDepth returns the number of Records 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 sane defaults.
type PayloadHook ¶
type PayloadHook struct {
// contains filtered or unexported fields
}
PayloadHook is the buffered-path producer: a lifecycle.Hook that builds a Record from the request body retained on the Context and the response body on the PostFlightEvent. Returns nil (nothing attached) when the request isn't opted into capture, so non-logged requests cost nothing beyond the gate check.
The Controller supplies the live master switch + per-body cap, both runtime-mutable via settings.
func NewPayloadHook ¶
func NewPayloadHook(c *Controller) *PayloadHook
NewPayloadHook constructs the producer bound to the Controller.
func (*PayloadHook) Fill ¶
func (h *PayloadHook) Fill(lc *lifecycle.Context, ev *lifecycle.PostFlightEvent) (any, error)
func (*PayloadHook) Name ¶
func (*PayloadHook) Name() string
type Reader ¶
Reader is the read-side contract (Get by request_id), re-exported so the control plane imports just payloadlog.
type Record ¶
Record / Sink / Closer are re-exported from pkg/payload so the observer and its wiring can import just payloadlog. The contract + backends live in pkg/payload and pkg/payload/<backend> (file, s3) — vendorable, and the s3 driver is excludable from minimal builds.
type SettingsSource ¶
type SettingsSource interface {
Setting(section string) (any, bool)
OnSettingsChange(section string, fn func())
}
SettingsSource reads a live settings section and subscribes to its changes. Satisfied by *app/catalog.Catalog.
type Sink ¶
Record / Sink / Closer are re-exported from pkg/payload so the observer and its wiring can import just payloadlog. The contract + backends live in pkg/payload and pkg/payload/<backend> (file, s3) — vendorable, and the s3 driver is excludable from minimal builds.
type SinkBuilder ¶
SinkBuilder constructs the concrete Sink for a resolved config. Injected by the composition root so the s3 driver stays behind a build tag — a minimal build supplies a builder that errors on backend "s3".
type SinkCollector ¶
type SinkCollector struct {
// contains filtered or unexported fields
}
SinkCollector is the payload janitor: a lifecycle.Collector that reads the *Record attached under Namespace and pushes it onto the bounded Emitter. Read-only on the Context; the Emit is a non-blocking channel send. No-op when no Record was attached (capture disabled for the request).
func NewSinkCollector ¶
func NewSinkCollector(e *Emitter) *SinkCollector
NewSinkCollector constructs a collector that emits collected Records onto e.
func (*SinkCollector) Collect ¶
func (c *SinkCollector) Collect(lc *lifecycle.Context)
type StreamPayloadFactory ¶
type StreamPayloadFactory struct {
// contains filtered or unexported fields
}
StreamPayloadFactory is the streamed-path counterpart to PayloadHook. Its observer accumulates the raw upstream SSE frames and builds the Record at end-of-stream (the buffered hook is too late — it runs after the stream is written). When capture is disabled for the request it returns a no-op observer so non-logged streams never buffer.
func NewStreamPayloadFactory ¶
func NewStreamPayloadFactory(c *Controller) *StreamPayloadFactory
NewStreamPayloadFactory constructs the factory bound to the Controller.
func (*StreamPayloadFactory) Name ¶
func (*StreamPayloadFactory) Name() string
func (*StreamPayloadFactory) NewObserver ¶
func (f *StreamPayloadFactory) NewObserver(lc *lifecycle.Context) lifecycle.StreamObserver
Source Files