nsq_consumer

README

NSQ Consumer Input Plugin

The NSQ consumer plugin reads from NSQD and creates metrics using one of the supported input data formats.

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 NSQD topic(s)
[[inputs.nsq_consumer]]
  ## An array representing the NSQD TCP HTTP Endpoints
  nsqd = ["localhost:4150"]

  ## An array representing the NSQLookupd HTTP Endpoints
  nsqlookupd = ["localhost:4161"]
  topic = "telegraf"
  channel = "consumer"
  max_in_flight = 100

  ## Max undelivered messages
  ## This plugin uses tracking metrics, which ensure messages are read to
  ## outputs before acknowledging them to the original broker to ensure data
  ## is not lost. This option sets the maximum messages to read from the
  ## broker that have not been written by an output.
  ##
  ## This value needs to be picked with awareness of the agent's
  ## metric_batch_size value as well. Setting max undelivered messages too high
  ## can result in a constant stream of data batches to the output. While
  ## setting it too low may never flush the broker's messages.
  # max_undelivered_messages = 1000

  ## 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"

Metrics

Example Output

Documentation

Index

• type NSQConsumer
  • func (n *NSQConsumer) Gather(_ telegraf.Accumulator) error
  • func (n *NSQConsumer) Init() error
  • func (*NSQConsumer) SampleConfig() string
  • func (n *NSQConsumer) SetParser(parser telegraf.Parser)
  • func (n *NSQConsumer) Start(ac telegraf.Accumulator) error
  • func (n *NSQConsumer) Stop()

Types

type NSQConsumer

type NSQConsumer struct {
	Server      string   `toml:"server" deprecated:"1.5.0;1.35.0;use 'nsqd' instead"`
	Nsqd        []string `toml:"nsqd"`
	Nsqlookupd  []string `toml:"nsqlookupd"`
	Topic       string   `toml:"topic"`
	Channel     string   `toml:"channel"`
	MaxInFlight int      `toml:"max_in_flight"`

	MaxUndeliveredMessages int `toml:"max_undelivered_messages"`

	Log telegraf.Logger
	// contains filtered or unexported fields
}

NSQConsumer represents the configuration of the plugin

func (*NSQConsumer) Gather

func (n *NSQConsumer) Gather(_ telegraf.Accumulator) error

Gather is a noop

func (*NSQConsumer) Init

func (n *NSQConsumer) Init() error

func (*NSQConsumer) SampleConfig

func (*NSQConsumer) SampleConfig() string

func (*NSQConsumer) SetParser

func (n *NSQConsumer) SetParser(parser telegraf.Parser)

SetParser takes the data_format from the config and finds the right parser for that format

func (*NSQConsumer) Start

func (n *NSQConsumer) Start(ac telegraf.Accumulator) error

Start pulls data from nsq

func (*NSQConsumer) Stop

func (n *NSQConsumer) Stop()

Stop processing messages