Rework mqtt_consumer connect/reconnect

[danielnelson]

Reworks the connection lifetime for the mqtt_consumer plugin. For now I have this marked for 1.9.0 but if we can get some help testing we can consider it for 1.8.x.

closes #4594
closes #4580
closes #4731

Based on #4814

Required for all PRs:

• Signed CLA.
• Associated README.md updated.
• Has appropriate unit tests.

[danielnelson]

• telegraf-1.9.0~dcf4fb4b-0.x86_64.rpm
• telegraf_1.9.0~dcf4fb4b-0_amd64.deb
• telegraf-1.9.0~dcf4fb4b_windows_amd64.zip

[trankennykhang]

@danielnelson sorry for slow action. It is the first time I contribute to community and just learned golang few weeks ago so I need a bit of time to check one by one carefully

Your changes is pretty good. Here is my comment:

• Three connection state will prevent the connection to establish second time completely. My change only tried to set it asap.
• Get rid of "receive" thread function to reduce the complexion
• I feel not right at the first look "opts.SetAutoReconnect(false)" but it is clear after checking further. The "SetConnectionLostHandler" will set the state back to original then the next call of "gather" function will try to establish it again.

I will test it on our IoTs project to see how it is. Thanks

[Dees7]

Please make armhf build for RaspberryPi3

[danielnelson]

• telegraf-1.9.0~dcf4fb4b-0.armv6hl.rpm
• telegraf-1.9.0~dcf4fb4b_linux_armhf.tar.gz
• telegraf_1.9.0~dcf4fb4b-0_armhf.deb

[Dees7]

There is to output at all.
telegraftest.conf :

[global_tags]
[agent]
  interval = "5s"
  round_interval = true
  metric_batch_size = 500
  metric_buffer_limit = 1000
  collection_jitter = "0s"
  flush_interval = "10s"
  flush_jitter = "0s"
  precision = ""
  debug = true
  quiet = false
  logfile = ""
  hostname = ""
  omit_hostname = true
[[processors.converter]]
  [processors.converter.tags]
    float = ["POWER*"]
[[processors.printer]]
  order = 2
[[processors.regex]]
  order = 1
  [[processors.regex.tags]]
      key = "topic"
      pattern = ".*/(.*)/.*"
      replacement = "${1}"
      result_key = "topicname"
[[inputs.mqtt_consumer]]
  servers = ["tcp://localhost:1883"]
  qos = 0
  connection_timeout = "4s"
  topics = ["test/#"]
  persistent_session = true
  client_id = "telegraftest"
  username = "user"
  password = "pass"
  name_override = "mqttc"
  tag_keys = ["hostname","POWER","POWER1"]
  data_format = "json"
[[outputs.file]]
   files = ["stdout"]
   data_format = "graphite"
[[outputs.influxdb]]
  urls = ["http://127.0.0.1:8086"]
  database = "test"
  skip_database_creation = false
  retention_policy = ""
  write_consistency = "any"
  timeout = "5s"
  username = "user"
  password = "pss"

/home/pi/telegraf --debug -config /etc/telegraf/telegraftest.conf

2018-10-11T06:51:26Z D! Attempting connection to output: influxdb
2018-10-11T06:51:26Z D! Successfully connected to output: influxdb
2018-10-11T06:51:26Z D! Attempting connection to output: file
2018-10-11T06:51:26Z D! Successfully connected to output: file
2018-10-11T06:51:26Z I! Starting Telegraf 
2018-10-11T06:51:26Z I! Loaded inputs: inputs.mqtt_consumer
2018-10-11T06:51:26Z I! Loaded aggregators: 
2018-10-11T06:51:26Z I! Loaded processors: converter regex printer
2018-10-11T06:51:26Z I! Loaded outputs: influxdb file
2018-10-11T06:51:26Z I! Tags enabled: 
2018-10-11T06:51:26Z I! Agent Config: Interval:5s, Quiet:false, Hostname:"", Flush Interval:10s 
2018-10-11T06:51:30Z I! [inputs.mqtt_consumer]: connected [tcp://localhost:1883]
2018-10-11T06:51:40Z D! Output [file] buffer fullness: 0 / 1000 metrics. 
2018-10-11T06:51:40Z D! Output [influxdb] buffer fullness: 0 / 1000 metrics. 
2018-10-11T06:51:50Z D! Output [influxdb] buffer fullness: 0 / 1000 metrics. 
2018-10-11T06:51:50Z D! Output [file] buffer fullness: 0 / 1000 metrics. 
^C2018-10-11T06:51:57Z I! Hang on, flushing any cached metrics before shutdown
2018-10-11T06:51:57Z D! Output [file] buffer fullness: 0 / 1000 metrics. 
2018-10-11T06:51:57Z D! Output [influxdb] buffer fullness: 0 / 1000 metrics. 
2018-10-11T06:51:57Z D! [inputs.mqtt_consumer]: disconnecting [tcp://localhost:1883]
2018-10-11T06:51:57Z D! [inputs.mqtt_consumer]: disconnected [tcp://localhost:1883]

publishing:

mosquitto_pub -u 'user' -P 'pass' -t 'test/sonoff/RESULT' -m '{"POWER":0,"TEMP":41.1,"hostname":"mine"}'
mosquitto_pub -u 'user' -P 'pass' -t 'test/sonoff/RESULT' -m '{"POWER":0,"TEMP":41.1,"hostname":"mine"}'

debug with mosquitto_sub -u 'user' -P 'pass' -v -t 'test/#':

test/sonoff/RESULT {"POWER":0,"TEMP":41.1,"hostname":"mine"}                                       
test/sonoff/RESULT {"POWER":0,"TEMP":41.1,"hostname":"mine"}  

mosquitto.log

Thu 11 Oct 11:51:30 +05 2018 1539240690: Sending CONNACK to telegraftest (1, 0)
Thu 11 Oct 11:51:34 +05 2018 1539240694: Sending PUBLISH to telegraftest (d0, q0, r0, m0, 'test/sonoff/RESULT', ... (41 bytes))
Thu 11 Oct 11:51:47 +05 2018 1539240707: Sending PUBLISH to telegraftest (d0, q0, r0, m0, 'test/sonoff/RESULT', ... (41 bytes))
Thu 11 Oct 11:51:57 +05 2018 1539240717: Received DISCONNECT from telegraftest

[danielnelson]

@Dees7 I had broken persistent sessions, but I have restored the behavior of subscribing once on the initial connection as we did before. I feel like this is not completely correct but I don't see a way to check the session present flag, I'll have to ask on the paho project but for now this should generally work.

• telegraf-1.9.0%7E2cc071d4_freebsd_amd64.tar.gz
• telegraf-1.9.0%7E2cc071d4_freebsd_i386.tar.gz
• telegraf-1.9.0%7E2cc071d4-0.x86_64.rpm
• telegraf-1.9.0%7E2cc071d4_linux_amd64.tar.gz
• telegraf_1.9.0%7E2cc071d4-0_amd64.deb
• telegraf-1.9.0%7E2cc071d4-0.arm64.rpm
• telegraf-1.9.0%7E2cc071d4_linux_arm64.tar.gz
• telegraf_1.9.0%7E2cc071d4-0_arm64.deb
• telegraf-1.9.0%7E2cc071d4-0.armel.rpm
• telegraf-1.9.0%7E2cc071d4_linux_armel.tar.gz
• telegraf_1.9.0%7E2cc071d4-0_armel.deb
• telegraf-1.9.0%7E2cc071d4-0.armv6hl.rpm
• telegraf-1.9.0%7E2cc071d4_linux_armhf.tar.gz
• telegraf_1.9.0%7E2cc071d4-0_armhf.deb
• telegraf-1.9.0%7E2cc071d4-0.i386.rpm
• telegraf-1.9.0%7E2cc071d4_linux_i386.tar.gz
• telegraf_1.9.0%7E2cc071d4-0_i386.deb
• telegraf-1.9.0%7E2cc071d4-0.s390x.rpm
• telegraf-1.9.0%7E2cc071d4_linux_s390x.tar.gz
• telegraf_1.9.0%7E2cc071d4-0_s390x.deb
• telegraf-1.9.0%7E2cc071d4-static_linux_amd64.tar.gz
• telegraf-1.9.0%7E2cc071d4_windows_amd64.zip
• telegraf-1.9.0%7E2cc071d4_windows_i386.zip

[chumbert2]

Using telegraf_1.9.0~dcf4fb4b-0_amd64.deb the problem of double connection we had with 1.8.0 and 1.8.1 is not present. MQTT messages are correctly received. (we haven't tested reconnect yet).

[Dees7]

Yes. Now works stable (after restarting telegraf or mosquitto).

[chumbert2]

Works as expected with telegraf_1.9.0~2cc071d4-0_amd64.deb:

• messages are properly received
• no double connection
• reconnection OK

NOTE: tested with persistent_session = false only (AWS broker does not support persistent sessions).

[Flowm]

Works perfectly so far on a RPi3 with telegraf_1.9.0%7E2cc071d4-0_armhf.deb
Thanks a lot!

[ErikAndren]

telegraf_1.9.0%7E2cc071d4-0_armhf.deb works for me on a raspberry pi 1 using debian stretch