Rework mqtt_consumer connect/reconnect #4846

danielnelson · 2018-10-10T22:50:55Z

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 · 2018-10-10T23:06:59Z

trankennykhang · 2018-10-11T00:39:47Z

@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 · 2018-10-11T04:57:12Z

Please make armhf build for RaspberryPi3

danielnelson · 2018-10-11T06:06:43Z

Dees7 · 2018-10-11T06:56:24Z

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 · 2018-10-11T08:54:14Z

@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.

chumbert2 · 2018-10-11T09:05:18Z

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 · 2018-10-11T09:10:25Z

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

chumbert2 · 2018-10-11T10:07:34Z

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 · 2018-10-12T23:53:17Z

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

ErikAndren · 2018-10-13T08:23:40Z

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

(cherry picked from commit 152365a)

danielnelson added 2 commits October 10, 2018 15:35

Fix race condition in mqtt_consumer connection

a59b573

Update mqtt_consumer log message style

dcf4fb4

danielnelson added the area/mqtt label Oct 10, 2018

danielnelson added this to the 1.9.0 milestone Oct 10, 2018

This was referenced Oct 10, 2018

MQTT input plugin disconnecting from Mosquitto #4594

Closed

MQTT connected, not pushing data to influxdb #4580

Closed

telegraf stops sending data to influx #4731

Closed

Prevent the mqtt establish the second connection (#4594) #4814

Closed

danielnelson added the fix pr to fix corresponding bug label Oct 10, 2018

danielnelson requested a review from glinton October 10, 2018 23:12

danielnelson mentioned this pull request Oct 10, 2018

strconv.ParseFloat: parsing "???": invalid syntax #4827

Closed

Subscribe to topics on first connection with persistent session

2cc071d

glinton approved these changes Oct 11, 2018

View reviewed changes

danielnelson modified the milestones: 1.9.0, 1.8.2 Oct 15, 2018

danielnelson merged commit 152365a into master Oct 15, 2018

danielnelson deleted the mqtt-connect branch October 15, 2018 20:03

danielnelson added a commit that referenced this pull request Oct 15, 2018

Rework mqtt_consumer connect/reconnect (#4846)

34926b4

(cherry picked from commit 152365a)

rgitzel pushed a commit to rgitzel/telegraf that referenced this pull request Oct 17, 2018

Rework mqtt_consumer connect/reconnect (influxdata#4846)

4c683b1

Terkwood mentioned this pull request Dec 4, 2018

Data stops flowing into Influx Terkwood/prawnalith#10

Closed

2 tasks

otherpirate pushed a commit to otherpirate/telegraf that referenced this pull request Mar 15, 2019

Rework mqtt_consumer connect/reconnect (influxdata#4846)

87176d3

otherpirate pushed a commit to otherpirate/telegraf that referenced this pull request Mar 15, 2019

Rework mqtt_consumer connect/reconnect (influxdata#4846)

b6b40f7

dupondje pushed a commit to dupondje/telegraf that referenced this pull request Apr 22, 2019

Rework mqtt_consumer connect/reconnect (influxdata#4846)

c2279b8

athoune pushed a commit to bearstech/telegraf that referenced this pull request Apr 17, 2020

Rework mqtt_consumer connect/reconnect (influxdata#4846)

d0b7248

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Rework mqtt_consumer connect/reconnect #4846

Rework mqtt_consumer connect/reconnect #4846

danielnelson commented Oct 10, 2018

danielnelson commented Oct 10, 2018

trankennykhang commented Oct 11, 2018

Dees7 commented Oct 11, 2018

danielnelson commented Oct 11, 2018

Dees7 commented Oct 11, 2018 •

edited

Loading

danielnelson commented Oct 11, 2018 •

edited

Loading

chumbert2 commented Oct 11, 2018

Dees7 commented Oct 11, 2018

chumbert2 commented Oct 11, 2018

Flowm commented Oct 12, 2018

ErikAndren commented Oct 13, 2018

Rework mqtt_consumer connect/reconnect #4846

Rework mqtt_consumer connect/reconnect #4846

Conversation

danielnelson commented Oct 10, 2018

Required for all PRs:

danielnelson commented Oct 10, 2018

trankennykhang commented Oct 11, 2018

Dees7 commented Oct 11, 2018

danielnelson commented Oct 11, 2018

Dees7 commented Oct 11, 2018 • edited Loading

danielnelson commented Oct 11, 2018 • edited Loading

chumbert2 commented Oct 11, 2018

Dees7 commented Oct 11, 2018

chumbert2 commented Oct 11, 2018

Flowm commented Oct 12, 2018

ErikAndren commented Oct 13, 2018

Dees7 commented Oct 11, 2018 •

edited

Loading

danielnelson commented Oct 11, 2018 •

edited

Loading