kafkaexporter

README

Kafka Exporter

Kafka exporter exports traces to Kafka. This exporter uses a synchronous producer that blocks and does not batch messages, therefore it should be used with batch and queued retry processors for higher throughput and resiliency. Message payload encoding is configurable.

The following settings are required:

• protocol_version (no default): Kafka protocol version e.g. 2.0.0

The following settings can be optionally configured:

• brokers (default = localhost:9092): The list of kafka brokers
• topic (default = otlp_spans): The name of the kafka topic to export to
• encoding (default = otlp_proto): The encoding of the payload sent to kafka. Available encodings:
  • otlp_proto: the payload is serialized to ExportTraceServiceRequest.
  • jaeger_proto: the payload is serialized to a single Jaeger proto Span.
  • jaeger_json: the payload is serialized to a single Jaeger JSON Span using jsonpb.
• metadata
  • full (default = true): Whether to maintain a full set of metadata. When disabled the client does not make the initial request to broker at the startup.
  • retry
    • max (default = 3): The number of retries to get metadata
    • backoff (default = 250ms): How long to wait between metadata retries
• timeout (default = 5s): Is the timeout for every attempt to send data to the backend.
• 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 = 120s): Is the maximum amount of time spent trying to send a batch; ignored if enabled is false
• sending_queue
  • enabled (default = false)
  • num_consumers (default = 10): Number of consumers that dequeue batches; ignored if enabled is false
  • queue_size (default = 5000): Maximum number of batches kept in memory before dropping data; ignored if enabled is false; User should calculate this as num_seconds * requests_per_second 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.

Example configuration:

exporters:
  kafka:
    brokers:
      - localhost:9092
    protocol_version: 2.0.0

Documentation

Index

• func NewFactory(options ...FactoryOption) component.ExporterFactory
• type Config
• type FactoryOption
  • func WithAddMarshallers(encodingMarshaller map[string]Marshaller) FactoryOption
• type Marshaller
• type Message
• type Metadata
• type MetadataRetry

Functions

func NewFactory

func NewFactory(options ...FactoryOption) component.ExporterFactory

NewFactory creates Kafka exporter factory.

Types

type Config

type Config struct {
	configmodels.ExporterSettings  `mapstructure:",squash"`
	exporterhelper.TimeoutSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct.
	exporterhelper.QueueSettings   `mapstructure:"sending_queue"`
	exporterhelper.RetrySettings   `mapstructure:"retry_on_failure"`

	// The list of kafka brokers (default localhost:9092)
	Brokers []string `mapstructure:"brokers"`
	// Kafka protocol version
	ProtocolVersion string `mapstructure:"protocol_version"`
	// The name of the kafka topic to export to (default "otlp_spans")
	Topic string `mapstructure:"topic"`
	// Encoding of the messages (default "otlp_proto")
	Encoding string `mapstructure:"encoding"`

	// Metadata is the namespace for metadata management properties used by the
	// Client, and shared by the Producer/Consumer.
	Metadata Metadata `mapstructure:"metadata"`
}

Config defines configuration for Kafka exporter.

type FactoryOption

type FactoryOption func(factory *kafkaExporterFactory)

FactoryOption applies changes to kafkaExporterFactory.

func WithAddMarshallers

func WithAddMarshallers(encodingMarshaller map[string]Marshaller) FactoryOption

WithAddMarshallers adds marshallers.

type Marshaller

type Marshaller interface {
	// Marshal serializes spans into Messages
	Marshal(traces pdata.Traces) ([]Message, error)

	// Encoding returns encoding name
	Encoding() string
}

Marshaller marshals traces into Message array.

type Message

type Message struct {
	Value []byte
}

Message encapsulates Kafka's message payload.

type Metadata

type Metadata struct {
	// Whether to maintain a full set of metadata for all topics, or just
	// the minimal set that has been necessary so far. The full set is simpler
	// and usually more convenient, but can take up a substantial amount of
	// memory if you have many topics and partitions. Defaults to true.
	Full bool `mapstructure:"full"`

	// Retry configuration for metadata.
	// This configuration is useful to avoid race conditions when broker
	// is starting at the same time as collector.
	Retry MetadataRetry `mapstructure:"retry"`
}

Metadata defines configuration for retrieving metadata from the broker.

type MetadataRetry

type MetadataRetry struct {
	// The total number of times to retry a metadata request when the
	// cluster is in the middle of a leader election or at startup (default 3).
	Max int `mapstructure:"max"`
	// How long to wait for leader election to occur before retrying
	// (default 250ms). Similar to the JVM's `retry.backoff.ms`.
	Backoff time.Duration `mapstructure:"backoff"`
}

MetadataRetry defines retry configuration for Metadata.