mqtt_consumer

README

MQTT Consumer Input Plugin

The MQTT consumer plugin reads from specified MQTT topics and adds messages to InfluxDB. The plugin expects messages in the Telegraf Input Data Formats.

Configuration:

# Read metrics from MQTT topic(s)
[[inputs.mqtt_consumer]]
  servers = ["localhost:1883"]
  ## MQTT QoS, must be 0, 1, or 2
  qos = 0

  ## Topics to subscribe to
  topics = [
    "telegraf/host01/cpu",
    "telegraf/+/mem",
    "sensors/#",
  ]

  ## Maximum number of metrics to buffer between collection intervals
  metric_buffer = 100000

  ## username and password to connect MQTT server.
  # username = "telegraf"
  # password = "metricsmetricsmetricsmetrics"

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

  ## Data format to consume. This can be "json", "influx" or "graphite"
  ## Each data format has it's 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"

Tags:

• All measurements are tagged with the incoming topic, ie topic=telegraf/host01/cpu

Documentation

Index

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

Types

type MQTTConsumer

type MQTTConsumer struct {
	Servers  []string
	Topics   []string
	Username string
	Password string
	QoS      int `toml:"qos"`

	// Legacy metric buffer support
	MetricBuffer int

	PersistentSession bool
	ClientID          string `toml:"client_id"`

	// 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"`
	// Use SSL but skip chain & host verification
	InsecureSkipVerify bool

	sync.Mutex
	// contains filtered or unexported fields
}

func (*MQTTConsumer) Description

func (m *MQTTConsumer) Description() string

func (*MQTTConsumer) Gather

func (m *MQTTConsumer) Gather(acc telegraf.Accumulator) error

func (*MQTTConsumer) SampleConfig

func (m *MQTTConsumer) SampleConfig() string

func (*MQTTConsumer) SetParser

func (m *MQTTConsumer) SetParser(parser parsers.Parser)

func (*MQTTConsumer) Start

func (m *MQTTConsumer) Start(acc telegraf.Accumulator) error

func (*MQTTConsumer) Stop

func (m *MQTTConsumer) Stop()