batch

Documentation

Index

• Constants
• Variables
• type Batch
  • func New(opts ...Option) *Batch
  • func (b *Batch) AddRecord(raw []byte, key string) error
  • func (b *Batch) Chunk() (chunks [][]types.PutRecordsRequestEntry)
• type Encoder
  • func NewEncoder(named string, batchOptions ...Option) (Encoder, error)
• type Option
  • func WithCompressionType(compressionType string) Option
  • func WithMaxRecordSize(size int) Option
  • func WithMaxRecordsPerBatch(limit int) Option

Constants

const (
	MaxRecordSize     = 1 << 20 // 1MiB
	MaxBatchedRecords = 500
)

Variables

var (
	// ErrPartitionKeyLength is used when the given key exceeds the allowed kinesis limit of 256 characters
	ErrPartitionKeyLength = errors.New("partition key size is greater than 256 characters")
	// ErrRecordLength is used when attempted record results in a byte array greater than 1MiB
	ErrRecordLength = consumererror.NewPermanent(errors.New("record size is greater than 1 MiB"))
)

var (
	// ErrUnsupportedEncoding is used when the encoder type does not support the type of encoding
	ErrUnsupportedEncoding = errors.New("unsupported type to encode")
	// ErrUnknownExportEncoder is used when a named encoding doesn't not exist
	ErrUnknownExportEncoder = errors.New("unknown encoding export format")
)

Types

type Batch

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

func New

func New(opts ...Option) *Batch

func (*Batch) AddRecord

func (b *Batch) AddRecord(raw []byte, key string) error

func (*Batch) Chunk

func (b *Batch) Chunk() (chunks [][]types.PutRecordsRequestEntry)

Chunk breaks up the iternal queue into blocks that can be used to be written to he kinesis.PutRecords endpoint

type Encoder

type Encoder interface {
	Metrics(md pmetric.Metrics) (*Batch, error)

	Traces(td ptrace.Traces) (*Batch, error)

	Logs(ld plog.Logs) (*Batch, error)
}

Encoder transforms the internal pipeline format into a configurable format that is then used to export to kinesis.

func NewEncoder

func NewEncoder(named string, batchOptions ...Option) (Encoder, error)

type Option

type Option func(bt *Batch)

func WithCompressionType

func WithCompressionType(compressionType string) Option

func WithMaxRecordSize

func WithMaxRecordSize(size int) Option

func WithMaxRecordsPerBatch

func WithMaxRecordsPerBatch(limit int) Option