kafka_consumer

README

Kafka Consumer Input Plugin

The Kafka consumer plugin polls a specified Kafka topic and adds messages to InfluxDB. The plugin assumes messages follow the line protocol. Consumer Group is used to talk to the Kafka cluster so multiple instances of telegraf can read from the same topic in parallel.

For old kafka version (< 0.8), please use the kafka_consumer_legacy input plugin and use the old zookeeper connection method.

Configuration

# Read metrics from Kafka topic(s)
[[inputs.kafka_consumer]]
  ## topic(s) to consume
  topics = ["telegraf"]
  brokers = ["localhost:9092"]
  ## the name of the consumer group
  consumer_group = "telegraf_metrics_consumers"
  ## Offset (must be either "oldest" or "newest")
  offset = "oldest"

  ## Optional SSL Config
  # ssl_ca = "/etc/telegraf/ca.pem"
  # ssl_cert = "/etc/telegraf/cert.pem"
  # ssl_key = "/etc/telegraf/key.pem"
  ## Use SSL but skip chain & host verification
  # insecure_skip_verify = false

  ## Optional SASL Config
  # sasl_username = "kafka"
  # sasl_password = "secret"

  ## Data format to consume.
  ## Each data format has its own unique set of configuration options, read
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
  data_format = "influx"

  ## Maximum length of a message to consume, in bytes (default 0/unlimited);
  ## larger messages are dropped
  max_message_len = 65536

Testing

Running integration tests requires running Zookeeper & Kafka. See Makefile for kafka container command.

Documentation

Index

• type Kafka
  • func (k *Kafka) Description() string
  • func (k *Kafka) Gather(acc telegraf.Accumulator) error
  • func (k *Kafka) SampleConfig() string
  • func (k *Kafka) SetParser(parser parsers.Parser)
  • func (k *Kafka) Start(acc telegraf.Accumulator) error
  • func (k *Kafka) Stop()

Types

type Kafka

type Kafka struct {
	ConsumerGroup string
	Topics        []string
	Brokers       []string
	MaxMessageLen int

	Cluster *cluster.Consumer

	// Verify Kafka SSL Certificate
	InsecureSkipVerify bool
	// Path to CA file
	SSLCA string `toml:"ssl_ca"`
	// Path to host cert file
	SSLCert string `toml:"ssl_cert"`
	// Path to cert key file
	SSLKey string `toml:"ssl_key"`

	// SASL Username
	SASLUsername string `toml:"sasl_username"`
	// SASL Password
	SASLPassword string `toml:"sasl_password"`

	// Legacy metric buffer support
	MetricBuffer int
	// TODO remove PointBuffer, legacy support
	PointBuffer int

	Offset string

	sync.Mutex
	// contains filtered or unexported fields
}

func (*Kafka) Description

func (k *Kafka) Description() string

func (*Kafka) Gather

func (k *Kafka) Gather(acc telegraf.Accumulator) error

func (*Kafka) SampleConfig

func (k *Kafka) SampleConfig() string

func (*Kafka) SetParser

func (k *Kafka) SetParser(parser parsers.Parser)

func (*Kafka) Start

func (k *Kafka) Start(acc telegraf.Accumulator) error

func (*Kafka) Stop

func (k *Kafka) Stop()