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]]
  ## MQTT broker URLs to be used. The format should be scheme://host:port,
  ## schema can be tcp, ssl, or ws.
  servers = ["tcp://localhost:1883"]
  ## MQTT QoS, must be 0, 1, or 2
  qos = 0
  ## Connection timeout for initial connection in seconds
  connection_timeout = "30s"

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

  # if true, messages that can't be delivered while the subscriber is offline
  # will be delivered when it comes back (such as on service restart).
  # NOTE: if true, client_id MUST be set
  persistent_session = false
  # If empty, a random client ID will be generated.
  client_id = ""

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

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"`
	ConnectionTimeout internal.Duration `toml:"connection_timeout"`

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