Fix MQTT input exits if Broker is not available on startup (#3202)

(cherry picked from commit 5488f4b3ac)
This commit is contained in:
DanKans 2017-09-11 20:24:51 +01:00 committed by Daniel Nelson
parent 34b7a4c361
commit e9a12bb694
No known key found for this signature in database
GPG Key ID: CAAD59C9444F6155
3 changed files with 52 additions and 20 deletions

View File

@ -13,6 +13,8 @@ The plugin expects messages in the
servers = ["localhost:1883"] servers = ["localhost:1883"]
## MQTT QoS, must be 0, 1, or 2 ## MQTT QoS, must be 0, 1, or 2
qos = 0 qos = 0
## Connection timeout for initial connection in seconds
connection_timeout = 30
## Topics to subscribe to ## Topics to subscribe to
topics = [ topics = [

View File

@ -21,6 +21,7 @@ type MQTTConsumer struct {
Username string Username string
Password string Password string
QoS int `toml:"qos"` QoS int `toml:"qos"`
ConnectionTimeout internal.Duration `toml:"connection_timeout"`
parser parsers.Parser parser parsers.Parser
@ -48,13 +49,15 @@ type MQTTConsumer struct {
// keep the accumulator internally: // keep the accumulator internally:
acc telegraf.Accumulator acc telegraf.Accumulator
started bool connected bool
} }
var sampleConfig = ` var sampleConfig = `
servers = ["localhost:1883"] servers = ["localhost:1883"]
## MQTT QoS, must be 0, 1, or 2 ## MQTT QoS, must be 0, 1, or 2
qos = 0 qos = 0
## Connection timeout for initial connection in seconds
connection_timeout = 30
## Topics to subscribe to ## Topics to subscribe to
topics = [ topics = [
@ -103,7 +106,7 @@ func (m *MQTTConsumer) SetParser(parser parsers.Parser) {
func (m *MQTTConsumer) Start(acc telegraf.Accumulator) error { func (m *MQTTConsumer) Start(acc telegraf.Accumulator) error {
m.Lock() m.Lock()
defer m.Unlock() defer m.Unlock()
m.started = false m.connected = false
if m.PersistentSession && m.ClientID == "" { if m.PersistentSession && m.ClientID == "" {
return fmt.Errorf("ERROR MQTT Consumer: When using persistent_session" + return fmt.Errorf("ERROR MQTT Consumer: When using persistent_session" +
@ -115,26 +118,40 @@ func (m *MQTTConsumer) Start(acc telegraf.Accumulator) error {
return fmt.Errorf("MQTT Consumer, invalid QoS value: %d", m.QoS) return fmt.Errorf("MQTT Consumer, invalid QoS value: %d", m.QoS)
} }
if int(m.ConnectionTimeout.Duration) <= 0 {
return fmt.Errorf("MQTT Consumer, invalid connection_timeout value: %d", m.ConnectionTimeout)
}
opts, err := m.createOpts() opts, err := m.createOpts()
if err != nil { if err != nil {
return err return err
} }
m.client = mqtt.NewClient(opts) m.client = mqtt.NewClient(opts)
if token := m.client.Connect(); token.Wait() && token.Error() != nil {
return token.Error()
}
m.in = make(chan mqtt.Message, 1000) m.in = make(chan mqtt.Message, 1000)
m.done = make(chan struct{}) m.done = make(chan struct{})
m.connect()
return nil
}
func (m *MQTTConsumer) connect() error {
if token := m.client.Connect(); token.Wait() && token.Error() != nil {
err := token.Error()
log.Printf("D! MQTT Consumer, connection error - %v", err)
return err
}
go m.receiver() go m.receiver()
return nil return nil
} }
func (m *MQTTConsumer) onConnect(c mqtt.Client) { func (m *MQTTConsumer) onConnect(c mqtt.Client) {
log.Printf("I! MQTT Client Connected") log.Printf("I! MQTT Client Connected")
if !m.PersistentSession || !m.started { if !m.PersistentSession || !m.connected {
topics := make(map[string]byte) topics := make(map[string]byte)
for _, topic := range m.Topics { for _, topic := range m.Topics {
topics[topic] = byte(m.QoS) topics[topic] = byte(m.QoS)
@ -145,7 +162,7 @@ func (m *MQTTConsumer) onConnect(c mqtt.Client) {
m.acc.AddError(fmt.Errorf("E! MQTT Subscribe Error\ntopics: %s\nerror: %s", m.acc.AddError(fmt.Errorf("E! MQTT Subscribe Error\ntopics: %s\nerror: %s",
strings.Join(m.Topics[:], ","), subscribeToken.Error())) strings.Join(m.Topics[:], ","), subscribeToken.Error()))
} }
m.started = true m.connected = true
} }
return return
} }
@ -186,18 +203,27 @@ func (m *MQTTConsumer) recvMessage(_ mqtt.Client, msg mqtt.Message) {
func (m *MQTTConsumer) Stop() { func (m *MQTTConsumer) Stop() {
m.Lock() m.Lock()
defer m.Unlock() defer m.Unlock()
if m.connected {
close(m.done) close(m.done)
m.client.Disconnect(200) m.client.Disconnect(200)
m.started = false m.connected = false
}
} }
func (m *MQTTConsumer) Gather(acc telegraf.Accumulator) error { func (m *MQTTConsumer) Gather(acc telegraf.Accumulator) error {
if !m.connected {
m.connect()
}
return nil return nil
} }
func (m *MQTTConsumer) createOpts() (*mqtt.ClientOptions, error) { func (m *MQTTConsumer) createOpts() (*mqtt.ClientOptions, error) {
opts := mqtt.NewClientOptions() opts := mqtt.NewClientOptions()
opts.ConnectTimeout = m.ConnectionTimeout.Duration
if m.ClientID == "" { if m.ClientID == "" {
opts.SetClientID("Telegraf-Consumer-" + internal.RandomString(5)) opts.SetClientID("Telegraf-Consumer-" + internal.RandomString(5))
} else { } else {
@ -238,6 +264,7 @@ func (m *MQTTConsumer) createOpts() (*mqtt.ClientOptions, error) {
opts.SetCleanSession(!m.PersistentSession) opts.SetCleanSession(!m.PersistentSession)
opts.SetOnConnectHandler(m.onConnect) opts.SetOnConnectHandler(m.onConnect)
opts.SetConnectionLostHandler(m.onConnectionLost) opts.SetConnectionLostHandler(m.onConnectionLost)
return opts, nil return opts, nil
} }

View File

@ -26,7 +26,9 @@ func newTestMQTTConsumer() (*MQTTConsumer, chan mqtt.Message) {
Servers: []string{"localhost:1883"}, Servers: []string{"localhost:1883"},
in: in, in: in,
done: make(chan struct{}), done: make(chan struct{}),
connected: true,
} }
return n, in return n, in
} }
@ -131,6 +133,7 @@ func TestRunParserAndGather(t *testing.T) {
n, in := newTestMQTTConsumer() n, in := newTestMQTTConsumer()
acc := testutil.Accumulator{} acc := testutil.Accumulator{}
n.acc = &acc n.acc = &acc
defer close(n.done) defer close(n.done)
n.parser, _ = parsers.NewInfluxParser() n.parser, _ = parsers.NewInfluxParser()