processor

README

General Information

Processors are used at various stages of a pipeline. Generally, a processor pre-processes data before it is exported (e.g. modify attributes or sample) or helps ensure that data makes it through a pipeline successfully (e.g. batch/retry).

Some important aspects of pipelines and processors to be aware of:

• Recommended Processors
• Data Ownership
• Exclusive Ownership
• Shared Ownership
• Ordering Processors

Supported processors (sorted alphabetically):

• Batch Processor
• Memory Limiter Processor

The contrib repository has more processors that can be added to a custom build of the Collector.

Recommended Processors

By default, no processors are enabled. Depending on the data source, it may be recommended that multiple processors be enabled. Processors must be enabled for every data source: Not all processors support all data sources. In addition, it is important to note that the order of processors matters. The order in each section below is the best practice. Refer to the individual processor documentation for more information.

Traces

1. memory_limiter
2. any sampling processors
3. Any processor relying on sending source from Context (e.g. k8sattributes)
4. batch
5. any other processors

Metrics

1. memory_limiter
2. Any processor relying on sending source from Context (e.g. k8sattributes)
3. batch
4. any other processors

Data Ownership

The ownership of the pdata.Traces, pdata.Metrics and pdata.Logs data in a pipeline is passed as the data travels through the pipeline. The data is created by the receiver and then the ownership is passed to the first processor when ConsumeTraces/ConsumeMetrics/ConsumeLogs function is called.

Note: the receiver may be attached to multiple pipelines, in which case the same data will be passed to all attached pipelines via a data fan-out connector.

From data ownership perspective pipelines can work in 2 modes:

• Exclusive data ownership
• Shared data ownership

The mode is defined during startup based on data modification intent reported by the processors. The intent is reported by each processor via MutatesData field of the struct returned by Capabilities function. If any processor in the pipeline declares an intent to modify the data then that pipeline will work in exclusive ownership mode. In addition, any other pipeline that receives data from a receiver that is attached to a pipeline with exclusive ownership mode will be also operating in exclusive ownership mode.

Exclusive Ownership

In exclusive ownership mode the data is owned exclusively by a particular processor at a given moment of time, and the processor is free to modify the data it owns.

Exclusive ownership mode is only applicable for pipelines that receive data from the same receiver. If a pipeline is marked to be in exclusive ownership mode then any data received from a shared receiver will be cloned at the fan-out connector before passing further to each pipeline. This ensures that each pipeline has its own exclusive copy of data, and the data can be safely modified in the pipeline.

The exclusive ownership of data allows processors to freely modify the data while they own it (e.g. see attributesprocessor). The duration of ownership of the data by processor is from the beginning of ConsumeTraces/ConsumeMetrics/ConsumeLogs call until the processor calls the next processor's ConsumeTraces/ConsumeMetrics/ConsumeLogs function, which passes the ownership to the next processor. After that the processor must no longer read or write the data since it may be concurrently modified by the new owner.

Exclusive Ownership mode allows to easily implement processors that need to modify the data by simply declaring such intent.

Shared Ownership

In shared ownership mode no particular processor owns the data and no processor is allowed the modify the shared data.

In this mode no cloning is performed at the fan-out connector of receivers that are attached to multiple pipelines. In this case all such pipelines will see the same single shared copy of the data. Processors in pipelines operating in shared ownership mode are prohibited from modifying the original data that they receive via ConsumeTraces/ConsumeMetrics/ConsumeLogs call. Processors may only read the data but must not modify the data.

If the processor needs to modify the data while performing the processing but does not want to incur the cost of data cloning that Exclusive mode brings then the processor can declare that it does not modify the data and use any different technique that ensures original data is not modified. For example, the processor can implement copy-on-write approach for individual sub-parts of pdata.Traces/pdata.Metrics/pdata.Logs argument. Any approach that does not mutate the original pdata.Traces/pdata.Metrics/pdata.Logs is allowed.

If the processor uses such technique it should declare that it does not intend to modify the original data by setting MutatesData=false in its capabilities to avoid marking the pipeline for Exclusive ownership and to avoid the cost of data cloning described in Exclusive Ownership section.

Ordering Processors

The order processors are specified in a pipeline is important as this is the order in which each processor is applied to traces and metrics.

Documentation

Index

• func MakeFactoryMap(factories ...Factory) (map[component.Type]Factory, error)
• type Builder
  • func NewBuilder(cfgs map[component.ID]component.Config, factories map[component.Type]Factory) *Builder
  • func (b *Builder) CreateLogs(ctx context.Context, set CreateSettings, next consumer.Logs) (Logs, error)
  • func (b *Builder) CreateMetrics(ctx context.Context, set CreateSettings, next consumer.Metrics) (Metrics, error)
  • func (b *Builder) CreateTraces(ctx context.Context, set CreateSettings, next consumer.Traces) (Traces, error)
  • func (b *Builder) Factory(componentType component.Type) component.Factory
• type CreateLogsFunc
  • func (f CreateLogsFunc) CreateLogsProcessor(ctx context.Context, set CreateSettings, cfg component.Config, ...) (Logs, error)
• type CreateMetricsFunc
  • func (f CreateMetricsFunc) CreateMetricsProcessor(ctx context.Context, set CreateSettings, cfg component.Config, ...) (Metrics, error)
• type CreateSettings
• type CreateTracesFunc
  • func (f CreateTracesFunc) CreateTracesProcessor(ctx context.Context, set CreateSettings, cfg component.Config, ...) (Traces, error)
• type Factory
  • func NewFactory(cfgType component.Type, createDefaultConfig component.CreateDefaultConfigFunc, ...) Factory
• type FactoryOption
  • func WithLogs(createLogs CreateLogsFunc, sl component.StabilityLevel) FactoryOption
  • func WithMetrics(createMetrics CreateMetricsFunc, sl component.StabilityLevel) FactoryOption
  • func WithTraces(createTraces CreateTracesFunc, sl component.StabilityLevel) FactoryOption
• type Logs
• type Metrics
• type Traces

Functions

func MakeFactoryMap

func MakeFactoryMap(factories ...Factory) (map[component.Type]Factory, error)

MakeFactoryMap takes a list of factories and returns a map with Factory type as keys. It returns a non-nil error when there are factories with duplicate type.

Types

type Builder

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

Builder processor is a helper struct that given a set of Configs and Factories helps with creating processors.

func NewBuilder

func NewBuilder(cfgs map[component.ID]component.Config, factories map[component.Type]Factory) *Builder

NewBuilder creates a new processor.Builder to help with creating components form a set of configs and factories.

func (*Builder) CreateLogs

func (b *Builder) CreateLogs(ctx context.Context, set CreateSettings, next consumer.Logs) (Logs, error)

CreateLogs creates a Logs processor based on the settings and config.

func (*Builder) CreateMetrics

func (b *Builder) CreateMetrics(ctx context.Context, set CreateSettings, next consumer.Metrics) (Metrics, error)

CreateMetrics creates a Metrics processor based on the settings and config.

func (*Builder) CreateTraces

func (b *Builder) CreateTraces(ctx context.Context, set CreateSettings, next consumer.Traces) (Traces, error)

CreateTraces creates a Traces processor based on the settings and config.

func (*Builder) Factory

func (b *Builder) Factory(componentType component.Type) component.Factory

type CreateLogsFunc

type CreateLogsFunc func(context.Context, CreateSettings, component.Config, consumer.Logs) (Logs, error)

CreateLogsFunc is the equivalent of Factory.CreateLogs().

func (CreateLogsFunc) CreateLogsProcessor

func (f CreateLogsFunc) CreateLogsProcessor(
	ctx context.Context,
	set CreateSettings,
	cfg component.Config,
	nextConsumer consumer.Logs,
) (Logs, error)

CreateLogsProcessor implements Factory.CreateLogsProcessor().

type CreateMetricsFunc

type CreateMetricsFunc func(context.Context, CreateSettings, component.Config, consumer.Metrics) (Metrics, error)

CreateMetricsFunc is the equivalent of Factory.CreateMetrics().

func (CreateMetricsFunc) CreateMetricsProcessor

func (f CreateMetricsFunc) CreateMetricsProcessor(
	ctx context.Context,
	set CreateSettings,
	cfg component.Config,
	nextConsumer consumer.Metrics,
) (Metrics, error)

CreateMetricsProcessor implements Factory.CreateMetricsProcessor().

type CreateSettings

type CreateSettings struct {
	// ID returns the ID of the component that will be created.
	ID component.ID

	component.TelemetrySettings

	// BuildInfo can be used by components for informational purposes
	BuildInfo component.BuildInfo
}

CreateSettings is passed to Create* functions in Factory.

type CreateTracesFunc

type CreateTracesFunc func(context.Context, CreateSettings, component.Config, consumer.Traces) (Traces, error)

CreateTracesFunc is the equivalent of Factory.CreateTraces().

func (CreateTracesFunc) CreateTracesProcessor

func (f CreateTracesFunc) CreateTracesProcessor(
	ctx context.Context,
	set CreateSettings,
	cfg component.Config,
	nextConsumer consumer.Traces) (Traces, error)

CreateTracesProcessor implements Factory.CreateTracesProcessor().

type Factory

type Factory interface {
	component.Factory

	// CreateTracesProcessor creates a TracesProcessor based on this config.
	// If the processor type does not support tracing or if the config is not valid,
	// an error will be returned instead.
	CreateTracesProcessor(ctx context.Context, set CreateSettings, cfg component.Config, nextConsumer consumer.Traces) (Traces, error)

	// TracesProcessorStability gets the stability level of the TracesProcessor.
	TracesProcessorStability() component.StabilityLevel

	// CreateMetricsProcessor creates a MetricsProcessor based on this config.
	// If the processor type does not support metrics or if the config is not valid,
	// an error will be returned instead.
	CreateMetricsProcessor(ctx context.Context, set CreateSettings, cfg component.Config, nextConsumer consumer.Metrics) (Metrics, error)

	// MetricsProcessorStability gets the stability level of the MetricsProcessor.
	MetricsProcessorStability() component.StabilityLevel

	// CreateLogsProcessor creates a LogsProcessor based on the config.
	// If the processor type does not support logs or if the config is not valid,
	// an error will be returned instead.
	CreateLogsProcessor(ctx context.Context, set CreateSettings, cfg component.Config, nextConsumer consumer.Logs) (Logs, error)

	// LogsProcessorStability gets the stability level of the LogsProcessor.
	LogsProcessorStability() component.StabilityLevel
	// contains filtered or unexported methods
}

Factory is Factory interface for processors.

This interface cannot be directly implemented. Implementations must use the NewProcessorFactory to implement it.

func NewFactory

func NewFactory(cfgType component.Type, createDefaultConfig component.CreateDefaultConfigFunc, options ...FactoryOption) Factory

NewFactory returns a Factory.

type FactoryOption

type FactoryOption interface {
	// contains filtered or unexported methods
}

FactoryOption apply changes to Options.

func WithLogs

func WithLogs(createLogs CreateLogsFunc, sl component.StabilityLevel) FactoryOption

WithLogs overrides the default "error not supported" implementation for CreateLogs and the default "undefined" stability level.

func WithMetrics

func WithMetrics(createMetrics CreateMetricsFunc, sl component.StabilityLevel) FactoryOption

WithMetrics overrides the default "error not supported" implementation for CreateMetrics and the default "undefined" stability level.

func WithTraces

func WithTraces(createTraces CreateTracesFunc, sl component.StabilityLevel) FactoryOption

WithTraces overrides the default "error not supported" implementation for CreateTraces and the default "undefined" stability level.

type Logs

type Logs interface {
	component.Component
	consumer.Logs
}

Logs is a processor that can consume logs.

type Metrics

type Metrics interface {
	component.Component
	consumer.Metrics
}

Metrics is a processor that can consume metrics.

type Traces

type Traces interface {
	component.Component
	consumer.Traces
}

Traces is a processor that can consume traces.