MQTT messages lost

[ardiehl]

when i have more than 1 or 2 panels using the mqtt data source, i will loose some messages, see the log below.

How to reproduce:
i have created a dummy configuration with emModbus2Influx (emModbus2Influx --configfile=config_03.conf) that sends 3 mqtt topics once a second. All 3 includes a count value that is always equal for all 3 topics.

ad/test/Live1 ad/test/Live2 ad/test/temp

All updates are received by grafana-mqtt-datasource but not all are forwarded, see my comments in the logfile below marked with >>

`logger=datasources t=2022-11-16T16:27:13.675580327+01:00 level=debug msg="Querying for data source via SQL store" id=1 orgId=1
logger=query_data t=2022-11-16T16:27:13.675811647+01:00 level=debug msg="Processing metrics query" query="unsupported value type"
logger=secrets.kvstore t=2022-11-16T16:27:13.67597391+01:00 level=debug msg="got secret value" orgId=1 type=datasource namespace=MQTT
logger=accesscontrol.service t=2022-11-16T16:27:13.67613391+01:00 level=debug msg="using cached permissions" key=rbac-permissions-1-user-2
logger=query_data t=2022-11-16T16:27:13.676219901+01:00 level=debug msg="Processing metrics query" query="unsupported value type"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:13.676257927+01:00 level=debug msg="Subscribing to MQTT topic: ad/test/Live2"
logger=secrets.kvstore t=2022-11-16T16:27:13.67626031+01:00 level=debug msg="got secret value from cache" orgId=1 type=datasource namespace=MQTT
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:13.67639417+01:00 level=debug msg="Subscribing to MQTT topic: ad/test/Live1"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:13.676423149+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live2"
logger=accesscontrol.service t=2022-11-16T16:27:13.683297121+01:00 level=debug msg="using cached permissions" key=rbac-permissions-1-user-2
logger=query_data t=2022-11-16T16:27:13.683358611+01:00 level=debug msg="Processing metrics query" query="unsupported value type"
logger=secrets.kvstore t=2022-11-16T16:27:13.683406845+01:00 level=debug msg="got secret value from cache" orgId=1 type=datasource namespace=MQTT
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:13.683596016+01:00 level=debug msg="Subscribing to MQTT topic: ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:13.683750535+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live1"
logger=live t=2022-11-16T16:27:13.691270352+01:00 level=debug msg="Client wants to subscribe" user=2 client=f87862d2-1c6e-4bab-828d-45f60e9831bd channel=1/ds/WqLUxaI4k/ad/test/Live2
logger=live t=2022-11-16T16:27:13.691300881+01:00 level=debug msg="Found cached channel handler" channel=ds/WqLUxaI4k/ad/test/Live2
logger=secrets.kvstore t=2022-11-16T16:27:13.691394813+01:00 level=debug msg="got secret value from cache" orgId=1 type=datasource namespace=MQTT
logger=live.features t=2022-11-16T16:27:13.691626945+01:00 level=debug msg="Running a new unidirectional stream" path=ad/test/Live2
logger=live t=2022-11-16T16:27:13.691644867+01:00 level=debug msg="Client subscribed" user=2 client=f87862d2-1c6e-4bab-828d-45f60e9831bd channel=1/ds/WqLUxaI4k/ad/test/Live2
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:13.691852513+01:00 level=debug msg="Sending message to client for topic ad/test/Live2"
logger=live t=2022-11-16T16:27:13.707822959+01:00 level=debug msg="Client wants to subscribe" user=2 client=f87862d2-1c6e-4bab-828d-45f60e9831bd channel=1/ds/WqLUxaI4k/ad/test/Live1
logger=live t=2022-11-16T16:27:13.707851169+01:00 level=debug msg="Found cached channel handler" channel=ds/WqLUxaI4k/ad/test/Live1
logger=secrets.kvstore t=2022-11-16T16:27:13.707949187+01:00 level=debug msg="got secret value from cache" orgId=1 type=datasource namespace=MQTT
logger=live.features t=2022-11-16T16:27:13.708108432+01:00 level=debug msg="Running a new unidirectional stream" path=ad/test/Live1
logger=live t=2022-11-16T16:27:13.70812821+01:00 level=debug msg="Client subscribed" user=2 client=f87862d2-1c6e-4bab-828d-45f60e9831bd channel=1/ds/WqLUxaI4k/ad/test/Live1
logger=live t=2022-11-16T16:27:13.721369874+01:00 level=debug msg="Client wants to subscribe" user=2 client=f87862d2-1c6e-4bab-828d-45f60e9831bd channel=1/ds/WqLUxaI4k/ad/test/temp
logger=live t=2022-11-16T16:27:13.721418496+01:00 level=debug msg="Found cached channel handler" channel=ds/WqLUxaI4k/ad/test/temp
logger=secrets.kvstore t=2022-11-16T16:27:13.721577594+01:00 level=debug msg="got secret value from cache" orgId=1 type=datasource namespace=MQTT
logger=live.features t=2022-11-16T16:27:13.721825165+01:00 level=debug msg="Running a new unidirectional stream" path=ad/test/temp
logger=live t=2022-11-16T16:27:13.721871792+01:00 level=debug msg="Client subscribed" user=2 client=f87862d2-1c6e-4bab-828d-45f60e9831bd channel=1/ds/WqLUxaI4k/ad/test/temp

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:13.724357377+01:00 level=debug msg="Received MQTT Message for topic ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:14.33442008+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live1"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:14.334460529+01:00 level=debug msg="Sending message to client for topic ad/test/Live1"
>>Received temp and Live1 but sending Live1 only

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:14.337465347+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live2"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:14.337488165+01:00 level=debug msg="Received MQTT Message for topic ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:15.338292212+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live1"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:15.338522928+01:00 level=debug msg="Sending message to client for topic ad/test/Live1"
>>Received 3 but send only Live1

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:15.34111404+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live2"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:15.341272993+01:00 level=debug msg="Received MQTT Message for topic ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:15.341347859+01:00 level=debug msg="Sending message to client for topic ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:15.341410063+01:00 level=debug msg="Sending message to client for topic ad/test/Live2"
>>ok

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:16.340872849+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live1"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:16.341019543+01:00 level=debug msg="Sending message to client for topic ad/test/Live1"
>>ok

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:16.343194401+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live2"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:16.343324883+01:00 level=debug msg="Received MQTT Message for topic ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:17.343235473+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live1"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:17.343272471+01:00 level=debug msg="Sending message to client for topic ad/test/Live1"
>>2 missing

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:17.346370486+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live2"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:17.346419652+01:00 level=debug msg="Received MQTT Message for topic ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:18.347473028+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live1"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:18.347564976+01:00 level=debug msg="Sending message to client for topic ad/test/Live1"
>>2 missing

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:18.348677675+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live2"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:18.348752423+01:00 level=debug msg="Received MQTT Message for topic ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:18.348793068+01:00 level=debug msg="Sending message to client for topic ad/test/Live2"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:18.348824901+01:00 level=debug msg="Sending message to client for topic ad/test/temp"
>>ok

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:19.35297471+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live1"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:19.353043616+01:00 level=debug msg="Sending message to client for topic ad/test/Live1"

ok

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:19.35393365+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live2"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:19.353958364+01:00 level=debug msg="Sending message to client for topic ad/test/Live2"
>>ok

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:19.353971612+01:00 level=debug msg="Received MQTT Message for topic ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:19.353982034+01:00 level=debug msg="Sending message to client for topic ad/test/temp"
>>ok

logger=ngalert.scheduler t=2022-11-16T16:27:20.000199076+01:00 level=debug msg="Alert rules fetched" rulesCount=0 foldersCount=0
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:20.354546413+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live1"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:20.354597225+01:00 level=debug msg="Sending message to client for topic ad/test/Live1"
>>ok

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:20.355737634+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live2"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:20.355770107+01:00 level=debug msg="Received MQTT Message for topic ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:20.355789649+01:00 level=debug msg="Sending message to client for topic ad/test/Live2"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:20.355808337+01:00 level=debug msg="Sending message to client for topic ad/test/temp"
>>ok

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:21.359707188+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live1"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:21.359746331+01:00 level=debug msg="Sending message to client for topic ad/test/Live1"
>>ok

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:21.360872038+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live2"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:21.360893181+01:00 level=debug msg="Received MQTT Message for topic ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:21.360905592+01:00 level=debug msg="Sending message to client for topic ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:21.360917236+01:00 level=debug msg="Sending message to client for topic ad/test/Live2"
>>ok

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:22.36365329+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live1"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:22.363716+01:00 level=debug msg="Sending message to client for topic ad/test/Live1"
>>ok

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:22.367414092+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live2"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:22.36746368+01:00 level=debug msg="Received MQTT Message for topic ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:23.369098973+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live1"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:23.369278097+01:00 level=debug msg="Sending message to client for topic ad/test/Live1"
>>2 missing

logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:23.37020893+01:00 level=debug msg="Received MQTT Message for topic ad/test/Live2"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:23.370343106+01:00 level=debug msg="Received MQTT Message for topic ad/test/temp"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:23.370418303+01:00 level=debug msg="Sending message to client for topic ad/test/Live2"
logger=plugin.grafana-mqtt-datasource t=2022-11-16T16:27:23.37047651+01:00 level=debug msg="Sending message to client for topic ad/test/temp"
>>ok

`

[jholubec]

I have the same issue.