kafka_consumer_legacy

README

Kafka Consumer Legacy 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.

Configuration

# Read metrics from Kafka topic(s)
[[inputs.kafka_consumer]]
  ## topic(s) to consume
  topics = ["telegraf"]

  ## an array of Zookeeper connection strings
  zookeeper_peers = ["localhost:2181"]

  ## Zookeeper Chroot
  zookeeper_chroot = ""

  ## the name of the consumer group
  consumer_group = "telegraf_metrics_consumers"

  ## Offset (must be either "oldest" or "newest")
  offset = "oldest"

  ## 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
	MaxMessageLen   int
	ZookeeperPeers  []string
	ZookeeperChroot string
	Consumer        *consumergroup.ConsumerGroup

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

	Offset string

	Log telegraf.Logger

	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()