consumer

Documentation

Index

• type Consumer
  • func New[T any](name string, opts Options, flushFn func(context.Context, []T) error) *Consumer[T]
  • func NewSized[T any](name string, opts Options, sizeOf func(T) int, ...) *Consumer[T]
  • func (c *Consumer[T]) Accepted() int64
  • func (c *Consumer[T]) Add(item T) bool
  • func (c *Consumer[T]) BufferedBytes() int64
  • func (c *Consumer[T]) Capacity() int
  • func (c *Consumer[T]) Dropped() int64
  • func (c *Consumer[T]) QueueLen() int
  • func (c *Consumer[T]) Start(ctx context.Context)
  • func (c *Consumer[T]) Stop()
  • func (c *Consumer[T]) WriteFailed() int64
• type Options

Types

type Consumer

type Consumer[T any] struct {
	// contains filtered or unexported fields
}

func New

func New[T any](name string, opts Options, flushFn func(context.Context, []T) error) *Consumer[T]

func NewSized

func NewSized[T any](name string, opts Options, sizeOf func(T) int, flushFn func(context.Context, []T) error) *Consumer[T]

NewSized is New plus a per-item byte sizer, enabling the Options.ByteBudget cap (v0.8.355, HA audit 🟡#1). sizeOf estimates one item's in-memory footprint — cheap string-length sums, never allocations (it runs on the Add hot path). A nil sizeOf or a zero ByteBudget leaves the consumer byte-identical to New.

func (*Consumer[T]) Accepted

func (c *Consumer[T]) Accepted() int64

Accepted returns the cumulative count of items that were successfully queued. Sampled twice over a known interval to compute an ingest rate.

func (*Consumer[T]) Add

func (c *Consumer[T]) Add(item T) bool

Add enqueues an item. Returns false if the buffer is full — by item count OR byte budget (v0.8.355) — and counts the drop either way.

func (*Consumer[T]) BufferedBytes

func (c *Consumer[T]) BufferedBytes() int64

BufferedBytes returns the approximate bytes currently held across the consumer's pipeline (channel + batches + in-flight flushes), per the NewSized sizer. Always 0 when byte accounting is disabled (v0.8.355).

func (*Consumer[T]) Capacity

func (c *Consumer[T]) Capacity() int

func (*Consumer[T]) Dropped

func (c *Consumer[T]) Dropped() int64

func (*Consumer[T]) QueueLen

func (c *Consumer[T]) QueueLen() int

func (*Consumer[T]) Start

func (c *Consumer[T]) Start(ctx context.Context)

func (*Consumer[T]) Stop

func (c *Consumer[T]) Stop()

Stop waits for the consumer loop and all flushers to finish after context cancellation.

func (*Consumer[T]) WriteFailed

func (c *Consumer[T]) WriteFailed() int64

WriteFailed returns the cumulative count of items lost because the ClickHouse insert (flushFn) errored — the batch was discarded, not retried. Surfaced on /admin/stats as the "write-failed" data-loss class.

type Options

type Options struct {
	BatchSize     int
	BufferSize    int
	FlushInterval time.Duration
	// Workers is the number of parallel flushers consuming the
	// dispatch channel. Each worker calls flushFn independently so
	// a slow ClickHouse insert no longer back-pressures item
	// accumulation. Defaults to 1 when unset for back-compat.
	Workers int
	// FlushTimeout bounds each flushFn call (v0.8.336). The old
	// context.Background() assumed CH's server-side
	// max_execution_time=60 bounds the insert — but a wedged server
	// that ACCEPTS the connection and never answers is bounded only
	// by the driver's 300s default ReadTimeout, and shutdown (Stop
	// waits on flushers) inherited whichever was worse. Defaults 60s.
	FlushTimeout time.Duration
	// FlushRetryBase is the first retry backoff (v0.8.340, HA audit
	// H2); the second retry waits 4×. Zero = the 2s production
	// default. Exposed mainly so tests run in milliseconds.
	FlushRetryBase time.Duration
	// ByteBudget caps the APPROXIMATE bytes this consumer may hold
	// across its whole pipeline — channel backlog + accumulating
	// batch + dispatched batches + batches in-flight in flushFn
	// (v0.8.355, HA audit 🟡#1). BufferSize alone bounds ITEMS: a
	// Java fleet emitting 15-25KB stack-trace log bodies could park
	// multi-GB behind a stalled-but-alive ClickHouse (batches occupy
	// workers for minutes under the v0.8.340 retry semantics) and get
	// the pod OOMKilled — destroying ALL buffered signals, worse than
	// the counted drops this budget produces instead. 0 = disabled.
	// Only enforced when the consumer was built with NewSized; a nil
	// sizer leaves behavior byte-identical to pre-v0.8.355.
	ByteBudget int64
}