kafka_consumer_legacy

package
v0.0.0-...-6a73ad5 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: May 1, 2019 License: MIT Imports: 9 Imported by: 0

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.

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

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

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

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

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL