exporterhelper

README

Exporter Helper

This is a helper exporter that other exporters can depend on. Today, it primarily offers queued retry capabilities.

⚠ This exporter should not be added to a service pipeline.

Configuration

The following configuration options can be modified:

• 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.
• sending_queue
  • enabled (default = true)
  • num_consumers (default = 10): Number of consumers that dequeue batches; ignored if enabled is false
  • queue_size (default = 1000): Maximum number of batches kept in memory before dropping; ignored if enabled is false User should calculate this as num_seconds * requests_per_second / requests_per_batch where:
    • num_seconds is the number of seconds to buffer in case of a backend outage
    • requests_per_second is the average number of requests per seconds
    • requests_per_batch is the average number of requests per batch (if the batch processor is used, the metric send_batch_size can be used for estimation)
• 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.

                                                              ┌─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:
    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

• func NewLogsExporter(ctx context.Context, set exporter.Settings, cfg component.Config, ...) (exporter.Logs, error)
• func NewLogsRequestExporter(_ context.Context, set exporter.Settings, converter RequestFromLogsFunc, ...) (exporter.Logs, error)
• func NewMetricsExporter(ctx context.Context, set exporter.Settings, cfg component.Config, ...) (exporter.Metrics, error)
• func NewMetricsRequestExporter(_ context.Context, set exporter.Settings, converter RequestFromMetricsFunc, ...) (exporter.Metrics, error)
• func NewThrottleRetry(err error, delay time.Duration) error
• func NewTracesExporter(ctx context.Context, set exporter.Settings, cfg component.Config, ...) (exporter.Traces, error)
• func NewTracesRequestExporter(_ context.Context, set exporter.Settings, converter RequestFromTracesFunc, ...) (exporter.Traces, error)
• type BatcherOption
  • func WithRequestBatchFuncs(mf exporterbatcher.BatchMergeFunc[Request], ...) BatcherOption
• type Option
  • func WithBatcher(cfg exporterbatcher.Config, opts ...BatcherOption) Option
  • func WithCapabilities(capabilities consumer.Capabilities) Option
  • func WithQueue(config internal.QueueConfig) Option
  • func WithRequestQueue(cfg exporterqueue.Config, queueFactory exporterqueue.Factory[Request]) 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 QueueConfig
  • func NewDefaultQueueConfig() QueueConfig
• type QueueSettings
  • func NewDefaultQueueSettings() QueueSettings
• type Request
• type RequestErrorHandler
• type RequestFromLogsFunc
• type RequestFromMetricsFunc
• type RequestFromTracesFunc
• type TimeoutConfig
  • func NewDefaultTimeoutConfig() TimeoutConfig
• type TimeoutSettings
  • func NewDefaultTimeoutSettings() TimeoutSettings

Functions

func NewLogsExporter

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

NewLogsExporter creates an exporter.Logs that records observability metrics and wraps every request with a Span.

func NewLogsRequestExporter

func NewLogsRequestExporter(
	_ context.Context,
	set exporter.Settings,
	converter RequestFromLogsFunc,
	options ...Option,
) (exporter.Logs, error)

NewLogsRequestExporter creates new logs exporter based on custom LogsConverter and RequestSender. Experimental: This API is at the early stage of development and may change without backward compatibility until https://github.com/open-telemetry/opentelemetry-collector/issues/8122 is resolved.

func NewMetricsExporter

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

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

func NewMetricsRequestExporter

func NewMetricsRequestExporter(
	_ context.Context,
	set exporter.Settings,
	converter RequestFromMetricsFunc,
	options ...Option,
) (exporter.Metrics, error)

NewMetricsRequestExporter creates a new metrics exporter based on a custom MetricsConverter and RequestSender. Experimental: This API is at the early stage of development and may change without backward compatibility until https://github.com/open-telemetry/opentelemetry-collector/issues/8122 is resolved.

func NewThrottleRetry

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

NewThrottleRetry creates a new throttle retry error.

func NewTracesExporter

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

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

func NewTracesRequestExporter

func NewTracesRequestExporter(
	_ context.Context,
	set exporter.Settings,
	converter RequestFromTracesFunc,
	options ...Option,
) (exporter.Traces, error)

NewTracesRequestExporter creates a new traces exporter based on a custom TracesConverter and RequestSender. Experimental: This API is at the early stage of development and may change without backward compatibility until https://github.com/open-telemetry/opentelemetry-collector/issues/8122 is resolved.

Types

type BatcherOption

type BatcherOption = internal.BatcherOption

BatcherOption apply changes to batcher sender.

func WithRequestBatchFuncs

func WithRequestBatchFuncs(mf exporterbatcher.BatchMergeFunc[Request], msf exporterbatcher.BatchMergeSplitFunc[Request]) BatcherOption

WithRequestBatchFuncs sets the functions for merging and splitting batches for an exporter built for custom request types.

type Option

type Option = internal.Option

Option apply changes to BaseExporter.

func WithBatcher

func WithBatcher(cfg exporterbatcher.Config, opts ...BatcherOption) Option

WithBatcher enables batching for an exporter based on custom request types. For now, it can be used only with the New[Traces|Metrics|Logs]RequestExporter exporter helpers and WithRequestBatchFuncs provided. This API is at the early stage of development and may change without backward compatibility until https://github.com/open-telemetry/opentelemetry-collector/issues/8122 is resolved.

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 internal.QueueConfig) Option

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

func WithRequestQueue

func WithRequestQueue(cfg exporterqueue.Config, queueFactory exporterqueue.Factory[Request]) Option

WithRequestQueue enables queueing for an exporter. This option should be used with the new exporter helpers New[Traces|Metrics|Logs]RequestExporter. Experimental: This API is at the early stage of development and may change without backward compatibility until https://github.com/open-telemetry/opentelemetry-collector/issues/8122 is resolved.

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 TimeoutSettings for an exporter. The default TimeoutSettings is 5 seconds.

type QueueConfig

type QueueConfig = internal.QueueConfig

QueueConfig defines configuration for queueing batches before sending to the consumerSender.

func NewDefaultQueueConfig

func NewDefaultQueueConfig() QueueConfig

NewDefaultQueueConfig returns the default config for QueueConfig.

type QueueSettings

type QueueSettings = internal.QueueConfig

Deprecated: [v0.110.0] Use QueueConfig instead.

func NewDefaultQueueSettings

func NewDefaultQueueSettings() QueueSettings

Deprecated: [v0.110.0] Use NewDefaultQueueConfig instead.

type Request

type Request = internal.Request

Request represents a single request that can be sent to an external endpoint. Experimental: This API is at the early stage of development and may change without backward compatibility until https://github.com/open-telemetry/opentelemetry-collector/issues/8122 is resolved.

type RequestErrorHandler

type RequestErrorHandler = internal.RequestErrorHandler

RequestErrorHandler is an optional interface that can be implemented by Request to provide a way handle partial temporary failures. For example, if some items failed to process and can be retried, this interface allows to return a new Request that contains the items left to be sent. Otherwise, the original Request should be returned. If not implemented, the original Request will be returned assuming the error is applied to the whole Request. Experimental: This API is at the early stage of development and may change without backward compatibility until https://github.com/open-telemetry/opentelemetry-collector/issues/8122 is resolved.

type RequestFromLogsFunc

type RequestFromLogsFunc func(context.Context, plog.Logs) (Request, error)

RequestFromLogsFunc converts plog.Logs data into a user-defined request. Experimental: This API is at the early stage of development and may change without backward compatibility until https://github.com/open-telemetry/opentelemetry-collector/issues/8122 is resolved.

type RequestFromMetricsFunc

type RequestFromMetricsFunc func(context.Context, pmetric.Metrics) (Request, error)

RequestFromMetricsFunc converts pdata.Metrics into a user-defined request. Experimental: This API is at the early stage of development and may change without backward compatibility until https://github.com/open-telemetry/opentelemetry-collector/issues/8122 is resolved.

type RequestFromTracesFunc

type RequestFromTracesFunc func(context.Context, ptrace.Traces) (Request, error)

RequestFromTracesFunc converts ptrace.Traces into a user-defined Request. Experimental: This API is at the early stage of development and may change without backward compatibility until https://github.com/open-telemetry/opentelemetry-collector/issues/8122 is resolved.

type TimeoutConfig

type TimeoutConfig = internal.TimeoutConfig

func NewDefaultTimeoutConfig

func NewDefaultTimeoutConfig() TimeoutConfig

NewDefaultTimeoutConfig returns the default config for TimeoutConfig.

type TimeoutSettings

type TimeoutSettings = TimeoutConfig

Deprecated: [v0.110.0] Use TimeoutConfig instead.

func NewDefaultTimeoutSettings

func NewDefaultTimeoutSettings() TimeoutSettings

Deprecated: [v0.110.0] Use NewDefaultTimeoutConfig instead.