kafka_consumer_legacy

README

Kafka Consumer Legacy Input Plugin

Deprecated in version 1.4. Please use 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.

Service Input

This plugin is a service input. Normal plugins gather metrics determined by the interval setting. Service plugins start a service to listens and waits for metrics or events to occur. Service plugins have two key differences from normal plugins:

1. The global or plugin specific interval setting may not apply
2. The CLI options of --test, --test-wait, and --once may not produce output for this plugin

Global configuration options

In addition to the plugin-specific configuration settings, plugins support additional global and plugin configuration settings. These settings are used to modify metrics, tags, and field or create aliases and configure ordering, etc. See the CONFIGURATION.md for more details.

Configuration

# Read metrics from Kafka topic(s)
[[inputs.kafka_consumer_legacy]]
  ## 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.

Metrics

Example Output

Documentation

Index

• type Kafka
  • func (k *Kafka) Gather(_ telegraf.Accumulator) error
  • func (*Kafka) SampleConfig() string
  • func (k *Kafka) SetParser(parser telegraf.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) Gather

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

func (*Kafka) SampleConfig

func (*Kafka) SampleConfig() string

func (*Kafka) SetParser

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

func (*Kafka) Start

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

func (*Kafka) Stop

func (k *Kafka) Stop()