exporterhelper

README

Exporter Helper

This package provides reusable implementations of common capabilities for exporters. Currently, this includes queuing, batching, timeouts, and retries.

Configuration

The following configuration options can be modified:

Retry on Failure

• retry_on_failure
  • enabled (default = true)
  • initial_interval (default = 5s): Time to wait after the first failure before retrying; ignored if enabled is false
  • max_interval (default = 30s): Is the upper bound on backoff; ignored if enabled is false
  • max_elapsed_time (default = 300s): Is the maximum amount of time spent trying to send a batch; ignored if enabled is false. If set to 0, the retries are never stopped.
  • multiplier (default = 1.5): Factor by which the retry interval is multiplied on each attempt; ignored if enabled is false

Sending Queue

• sending_queue
  • enabled (default = true)
  • num_consumers (default = 10): Number of consumers that dequeue batches; ignored if enabled is false
  • wait_for_result (default = false): determines if incoming requests are blocked until the request is processed or not.
  • block_on_overflow (default = false): If true, blocks the request until the queue has space otherwise rejects the data immediately; ignored if enabled is false
  • sizer (default = requests): How the queue and batching is measured. Available options:
    • requests: number of incoming batches of metrics, logs, traces (the most performant option);
    • items: number of the smallest parts of each signal (spans, metric data points, log records);
    • bytes: the size of serialized data in bytes (the least performant option).
  • queue_size (default = 1000): Maximum size the queue can accept. Measured in units defined by sizer
  • batch: see below.

Failure behavior: If data cannot be added to the sending queue, it is typically dropped. This occurs when the queue has reached its configured capacity or, for persistent queues, when the underlying storage cannot accept additional data (for example, due to insufficient disk space or I/O errors).

When block_on_overflow is enabled, the caller may instead wait until space becomes available, and the request may still be enqueued if capacity frees up before the timeout.

If data is rejected before entering the queue, it does not reach the exporter retry logic. Such enqueue failures are reported by the otelcol_exporter_enqueue_failed_* metrics.

Sending queue batch settings

Batch settings are available in the sending queue. Batching is disabled, by default. To enable default batch settings, use batch: {}. When batch is defined, the settings are:

• flush_timeout (default = 200 ms): time after which a batch will be sent regardless of its size. Must be a non-zero value;
• min_size (default = 8192): the minimum size of a batch; should be less than or equal to the sending_queue::queue_size if sending_queue::batch::sizer matches sending_queue::sizer.
• max_size (default = 0): the maximum size of a batch, enables batch splitting. The maximum size of a batch should be greater than or equal to the minimum size of a batch. If set to zero, there is no maximum size;
• sizer: see below.
• partition: see below.

The batch::sizer field is given special treatment because the queue itself also defines a sizer. This field supports using different size limits for the queue and batch-related logic.

If the batch::sizer field is not set, it takes its value from the parent structure.

If sending_queue::sizer is not set, batch::sizer defaults to items.

Available batch::sizer options:

• items: number of the smallest parts of each signal (spans, metric data points, log records);
• bytes: the size of serialized data in bytes (the least performant option).

The batch::partition configuration defines the partitioning of the batches.

Available batch::partition options:

• metadata_keys: a list of client.Metadata keys used to partition data into separate batches. When empty, a single batcher instance is used. When set, one batcher will be used per distinct combination of values for the listed metadata keys. Empty value and unset metadata are treated as distinct cases. Entries are case-insensitive. Duplicated entries will trigger a validation error. Default is empty.
• cache_size (default = 10000): maximum number of active partition batchers kept in the LRU cache when partitioning is enabled. Must be positive. When the limit is reached, the least recently used partition is flushed and removed. The current size and configured capacity are exported as otelcol_exporter_queue_batch_partition_cache_size and otelcol_exporter_queue_batch_partition_cache_capacity.
• idle_timeout (default = 90s): how long a partition may stay empty before it is removed. Keep it above the data arrival interval to avoid churning partitions on every scrape. Must be positive.

Timeout

• timeout (default = 5s): Time to wait per individual attempt to send data to a backend

The initial_interval, max_interval, max_elapsed_time, and timeout options accept duration strings, valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".

Persistent Queue

To use the persistent queue, the following setting needs to be set:

• sending_queue
  • storage (default = none): When set, enables persistence and uses the component specified as a storage extension for the persistent queue. There is no in-memory queue when set.

The maximum number of batches stored to disk can be controlled using sending_queue.queue_size parameter (which, similarly as for in-memory buffering, defaults to 1000 batches).

When persistent queue is enabled, the batches are being buffered using the provided storage extension - filestorage is a popular and safe choice. If the collector instance is killed while having some items in the persistent queue, on restart the items will be picked and the exporting is continued.

Context Propagation: Request context (including client metadata and span context) is preserved when using persistent queues. However, context set by Auth extensions is not propagated through the persistent queue. Auth extension context is ignored when data is persisted to disk, which means authentication/authorization information will not be available when the persisted data is processed.

                                                              ┌─Consumer #1─┐
                                                              │    ┌───┐    │
                              ──────Deleted──────        ┌───►│    │ 1 │    ├───► Success
        Waiting in channel    x           x     x        │    │    └───┘    │
        for consumer ───┐     x           x     x        │    │             │
                        │     x           x     x        │    └─────────────┘
                        ▼     x           x     x        │
┌─────────────────────────────────────────x─────x───┐    │    ┌─Consumer #2─┐
│                             x           x     x   │    │    │    ┌───┐    │
│     ┌───┐     ┌───┐ ┌───┐ ┌─x─┐ ┌───┐ ┌─x─┐ ┌─x─┐ │    │    │    │ 2 │    ├───► Permanent -> X
│ n+1 │ n │ ... │ 6 │ │ 5 │ │ 4 │ │ 3 │ │ 2 │ │ 1 │ ├────┼───►│    └───┘    │      failure
│     └───┘     └───┘ └───┘ └───┘ └───┘ └───┘ └───┘ │    │    │             │
│                                                   │    │    └─────────────┘
└───────────────────────────────────────────────────┘    │
   ▲              ▲     ▲           ▲                    │    ┌─Consumer #3─┐
   │              │     │           │                    │    │    ┌───┐    │
   │              │     │           │                    │    │    │ 3 │    ├───► (in progress)
 write          read    └─────┬─────┘                    ├───►│    └───┘    │
 index          index         │                          │    │             │
                              │                          │    └─────────────┘
                              │                          │
                          currently                      │    ┌─Consumer #4─┐
                          dispatched                     │    │    ┌───┐    │     Temporary
                                                         └───►│    │ 4 │    ├───►  failure
                                                              │    └───┘    │         │
                                                              │             │         │
                                                              └─────────────┘         │
                                                                     ▲                │
                                                                     └── Retry ───────┤
                                                                                      │
                                                                                      │
                                                   X  ◄────── Retry limit exceeded ───┘

Example:

receivers:
  otlp:
    protocols:
      grpc:
exporters:
  otlp_grpc:
    endpoint: <ENDPOINT>
    sending_queue:
      storage: file_storage/otc
extensions:
  file_storage/otc:
    directory: /var/lib/storage/otc
    timeout: 10s
service:
  extensions: [file_storage]
  pipelines:
    metrics:
      receivers: [otlp]
      exporters: [otlp]
    logs:
      receivers: [otlp]
      exporters: [otlp]
    traces:
      receivers: [otlp]
      exporters: [otlp]

Documentation

Overview

Package exporterhelper provides helper functions for exporters.

Index

• Variables
• func NewLogs(ctx context.Context, set exporter.Settings, cfg component.Config, ...) (exporter.Logs, error)
• func NewMetrics(ctx context.Context, set exporter.Settings, cfg component.Config, ...) (exporter.Metrics, error)
• func NewThrottleRetry(err error, delay time.Duration) error
• func NewTraces(ctx context.Context, set exporter.Settings, cfg component.Config, ...) (exporter.Traces, error)
• type BatchConfig
• type Option
  • func WithAttrs(attrs ...attribute.KeyValue) Option
  • func WithCapabilities(capabilities consumer.Capabilities) Option
  • func WithQueue(config configoptional.Optional[QueueBatchConfig]) Option
  • func WithRetry(config configretry.BackOffConfig) Option
  • func WithShutdown(shutdown component.ShutdownFunc) Option
  • func WithStart(start component.StartFunc) Option
  • func WithTimeout(timeoutConfig TimeoutConfig) Option
• type QueueBatchConfig
• type QueueBatchEncoding
• type RequestSizerType
• type TimeoutConfig
  • func NewDefaultTimeoutConfig() TimeoutConfig

Variables

var (
	RequestSizerTypeBytes    = request.SizerTypeBytes
	RequestSizerTypeItems    = request.SizerTypeItems
	RequestSizerTypeRequests = request.SizerTypeRequests
)

var ErrQueueIsFull = queue.ErrQueueIsFull

var NewDefaultBatchConfig = queuebatch.NewDefaultBatchConfig

NewDefaultBatchConfig returns the default config for BatchConfig.

var NewDefaultQueueConfig = internal.NewDefaultQueueConfig

NewDefaultQueueConfig returns the default config for QueueBatchConfig. By default, the queue stores 1000 requests of telemetry and is non-blocking when full.

Functions

func NewLogs

func NewLogs(
	ctx context.Context,
	set exporter.Settings,
	cfg component.Config,
	pusher consumer.ConsumeLogsFunc,
	options ...Option,
) (exporter.Logs, error)

NewLogs creates an exporter.Logs that records observability logs and wraps every request with a Span.

func NewMetrics

func NewMetrics(
	ctx context.Context,
	set exporter.Settings,
	cfg component.Config,
	pusher consumer.ConsumeMetricsFunc,
	options ...Option,
) (exporter.Metrics, error)

NewMetrics creates an exporter.Metrics that records observability metrics and wraps every request with a Span.

func NewThrottleRetry

func NewThrottleRetry(err error, delay time.Duration) error

NewThrottleRetry creates a new throttle retry error.

func NewTraces

func NewTraces(
	ctx context.Context,
	set exporter.Settings,
	cfg component.Config,
	pusher consumer.ConsumeTracesFunc,
	options ...Option,
) (exporter.Traces, error)

NewTraces creates an exporter.Traces that records observability metrics and wraps every request with a Span.

Types

type BatchConfig

type BatchConfig = queuebatch.BatchConfig

BatchConfig defines a configuration for batching requests based on a timeout and a minimum number of items.

type Option

type Option = internal.Option

Option apply changes to BaseExporter.

func WithAttrs

func WithAttrs(attrs ...attribute.KeyValue) Option

WithAttrs adds extra attributes to the metrics produced by the exporter The default set of extra attribute is empty

func WithCapabilities

func WithCapabilities(capabilities consumer.Capabilities) Option

WithCapabilities overrides the default Capabilities() function for a Consumer. The default is non-mutable data. TODO: Verify if we can change the default to be mutable as we do for processors.

func WithQueue

func WithQueue(config configoptional.Optional[QueueBatchConfig]) Option

WithQueue overrides the default QueueBatchConfig for an exporter. The default QueueBatchConfig is to disable queueing. This option cannot be used with the new exporter helpers New[Traces|Metrics|Logs]RequestExporter.

func WithRetry

func WithRetry(config configretry.BackOffConfig) Option

WithRetry overrides the default configretry.BackOffConfig for an exporter. The default configretry.BackOffConfig is to disable retries.

func WithShutdown

func WithShutdown(shutdown component.ShutdownFunc) Option

WithShutdown overrides the default Shutdown function for an exporter. The default shutdown function does nothing and always returns nil.

func WithStart

func WithStart(start component.StartFunc) Option

WithStart overrides the default Start function for an exporter. The default start function does nothing and always returns nil.

func WithTimeout

func WithTimeout(timeoutConfig TimeoutConfig) Option

WithTimeout overrides the default TimeoutConfig for an exporter. The default TimeoutConfig is 5 seconds.

type QueueBatchConfig

type QueueBatchConfig = queuebatch.Config

QueueBatchConfig defines configuration for queueing and batching for the exporter.

type QueueBatchEncoding

type QueueBatchEncoding[T any] interface {
	// Marshal is a function that can marshal a request and its context into bytes.
	Marshal(context.Context, T) ([]byte, error)

	// Unmarshal is a function that can unmarshal bytes into a request and its context.
	Unmarshal([]byte) (context.Context, T, error)
}

QueueBatchEncoding defines the encoding to be used if persistent queue is configured. Duplicate definition with queuebatch.Encoding since aliasing generics is not supported by default.

type RequestSizerType

type RequestSizerType = request.SizerType

type TimeoutConfig

type TimeoutConfig = internal.TimeoutConfig

TimeoutConfig a timeout configuration. The timeout applies to individual attempts to send data to the backend.

func NewDefaultTimeoutConfig

func NewDefaultTimeoutConfig() TimeoutConfig

NewDefaultTimeoutConfig returns a new TimeoutConfig with its default values applied. Prefer this over an empty struct literal so default changes stay in sync automatically.