kafka

Documentation

Overview

Package kafka provides an otelcol.receiver.kafka component.

Index

• Variables
• type AWSMSKArguments
  • func (args AWSMSKArguments) Convert() kafkaexporter.AWSMSKConfig
• type Arguments
  • func (args Arguments) Convert() (otelconfig.Receiver, error)
  • func (args Arguments) Exporters() map[otelconfig.DataType]map[otelconfig.ComponentID]otelcomponent.Exporter
  • func (args Arguments) Extensions() map[otelconfig.ComponentID]otelcomponent.Extension
  • func (args Arguments) NextConsumers() *otelcol.ConsumerArguments
  • func (args *Arguments) UnmarshalRiver(f func(interface{}) error) error
• type AuthenticationArguments
  • func (args AuthenticationArguments) Convert() kafkaexporter.Authentication
• type AutoCommitArguments
  • func (args AutoCommitArguments) Convert() kafkareceiver.AutoCommit
• type KerberosArguments
  • func (args KerberosArguments) Convert() kafkaexporter.KerberosConfig
• type MessageMarkingArguments
  • func (args MessageMarkingArguments) Convert() kafkareceiver.MessageMarking
• type MetadataArguments
  • func (args MetadataArguments) Convert() kafkaexporter.Metadata
• type MetadataRetryArguments
  • func (args MetadataRetryArguments) Convert() kafkaexporter.MetadataRetry
• type PlaintextArguments
  • func (args PlaintextArguments) Convert() kafkaexporter.PlainTextConfig
• type SASLArguments
  • func (args SASLArguments) Convert() kafkaexporter.SASLConfig

Variables

var DefaultArguments = Arguments{

	Topic:    "otlp_spans",
	Encoding: "otlp_proto",
	Brokers:  []string{"localhost:9092"},
	ClientID: "otel-collector",
	GroupID:  "otel-collector",
	Metadata: MetadataArguments{
		IncludeAllTopics: true,
		Retry: MetadataRetryArguments{
			MaxRetries: 3,
			Backoff:    250 * time.Millisecond,
		},
	},
	AutoCommit: AutoCommitArguments{
		Enable:   true,
		Interval: time.Second,
	},
	MessageMarking: MessageMarkingArguments{
		AfterExecution:      false,
		IncludeUnsuccessful: false,
	},
}

DefaultArguments holds default values for Arguments.

Types

type AWSMSKArguments

type AWSMSKArguments struct {
	Region     string `river:"region,attr"`
	BrokerAddr string `river:"broker_addr,attr"`
}

AWSMSKArguments exposes additional SASL authentication measures required to use the AWS_MSK_IAM mechanism.

func (AWSMSKArguments) Convert

func (args AWSMSKArguments) Convert() kafkaexporter.AWSMSKConfig

Convert converts args into the upstream type.

type Arguments

type Arguments struct {
	Brokers         []string `river:"brokers,attr"`
	ProtocolVersion string   `river:"protocol_version,attr"`
	Topic           string   `river:"topic,attr,optional"`
	Encoding        string   `river:"encoding,attr,optional"`
	GroupID         string   `river:"group_id,attr,optional"`
	ClientID        string   `river:"client_id,attr,optional"`

	Authentication AuthenticationArguments `river:"authentication,block,optional"`
	Metadata       MetadataArguments       `river:"metadata,block,optional"`
	AutoCommit     AutoCommitArguments     `river:"autocommit,block,optional"`
	MessageMarking MessageMarkingArguments `river:"message_marking,block,optional"`

	// Output configures where to send received data. Required.
	Output *otelcol.ConsumerArguments `river:"output,block"`
}

Arguments configures the otelcol.receiver.kafka component.

func (Arguments) Convert

func (args Arguments) Convert() (otelconfig.Receiver, error)

Convert implements receiver.Arguments.

func (Arguments) Exporters

func (args Arguments) Exporters() map[otelconfig.DataType]map[otelconfig.ComponentID]otelcomponent.Exporter

Exporters implements receiver.Arguments.

func (Arguments) Extensions

func (args Arguments) Extensions() map[otelconfig.ComponentID]otelcomponent.Extension

Extensions implements receiver.Arguments.

func (Arguments) NextConsumers

func (args Arguments) NextConsumers() *otelcol.ConsumerArguments

NextConsumers implements receiver.Arguments.

func (*Arguments) UnmarshalRiver

func (args *Arguments) UnmarshalRiver(f func(interface{}) error) error

UnmarshalRiver implements river.Unmarshaler and applies default settings.

type AuthenticationArguments

type AuthenticationArguments struct {
	Plaintext *PlaintextArguments         `river:"plaintext,block,optional"`
	SASL      *SASLArguments              `river:"sasl,block,optional"`
	TLS       *otelcol.TLSClientArguments `river:"tls,block,optional"`
	Kerberos  *KerberosArguments          `river:"kerberos,block,optional"`
}

AuthenticationArguments configures how to authenticate to the Kafka broker.

func (AuthenticationArguments) Convert

func (args AuthenticationArguments) Convert() kafkaexporter.Authentication

Convert converts args into the upstream type.

type AutoCommitArguments

type AutoCommitArguments struct {
	Enable   bool          `river:"enable,attr,optional"`
	Interval time.Duration `river:"interval,attr,optional"`
}

AutoCommitArguments configures how to automatically commit updated topic offsets back to the Kafka broker.

func (AutoCommitArguments) Convert

func (args AutoCommitArguments) Convert() kafkareceiver.AutoCommit

Convert converts args into the upstream type.

type KerberosArguments

type KerberosArguments struct {
	ServiceName string            `river:"service_name,attr,optional"`
	Realm       string            `river:"realm,attr,optional"`
	UseKeyTab   bool              `river:"use_keytab,attr,optional"`
	Username    string            `river:"username,attr"`
	Password    rivertypes.Secret `river:"password,attr,optional"`
	ConfigPath  string            `river:"config_file,attr,optional"`
	KeyTabPath  string            `river:"keytab_file,attr,optional"`
}

KerberosArguments configures Kerberos authentication against the Kafka broker.

func (KerberosArguments) Convert

func (args KerberosArguments) Convert() kafkaexporter.KerberosConfig

Convert converts args into the upstream type.

type MessageMarkingArguments

type MessageMarkingArguments struct {
	AfterExecution      bool `river:"after_execution,attr,optional"`
	IncludeUnsuccessful bool `river:"include_unsuccessful,attr,optional"`
}

MessageMarkingArguments configures when Kafka messages are marked as read.

func (MessageMarkingArguments) Convert

func (args MessageMarkingArguments) Convert() kafkareceiver.MessageMarking

Convert converts args into the upstream type.

type MetadataArguments

type MetadataArguments struct {
	IncludeAllTopics bool                   `river:"include_all_topics,attr,optional"`
	Retry            MetadataRetryArguments `river:"retry,block,optional"`
}

MetadataArguments configures how the otelcol.receiver.kafka component will retrieve metadata from the Kafka broker.

func (MetadataArguments) Convert

func (args MetadataArguments) Convert() kafkaexporter.Metadata

Convert converts args into the upstream type.

type MetadataRetryArguments

type MetadataRetryArguments struct {
	MaxRetries int           `river:"max_retries,attr,optional"`
	Backoff    time.Duration `river:"backoff,attr,optional"`
}

MetadataRetryArguments configures how to retry retrieving metadata from the Kafka broker. Retrying is useful to avoid race conditions when the Kafka broker is starting at the same time as the otelcol.receiver.kafka component.

func (MetadataRetryArguments) Convert

func (args MetadataRetryArguments) Convert() kafkaexporter.MetadataRetry

Convert converts args into the upstream type.

type PlaintextArguments

type PlaintextArguments struct {
	Username string            `river:"username,attr"`
	Password rivertypes.Secret `river:"password,attr"`
}

PlaintextArguments configures plaintext authentication against the Kafka broker.

func (PlaintextArguments) Convert

func (args PlaintextArguments) Convert() kafkaexporter.PlainTextConfig

Convert converts args into the upstream type.

type SASLArguments

type SASLArguments struct {
	Username  string            `river:"username,attr"`
	Password  rivertypes.Secret `river:"password,attr"`
	Mechanism string            `river:"mechanism,attr"`
	AWSMSK    AWSMSKArguments   `river:"aws_msk,block,optional"`
}

SASLArguments configures SASL authentication against the Kafka broker.

func (SASLArguments) Convert

func (args SASLArguments) Convert() kafkaexporter.SASLConfig

Convert converts args into the upstream type.