Unreliable MQTT subscription on high data rate source

[joriwinderickx-fm]

I am evaluating the usage of a FIWARE setup in an "industrial environment". At the moment, I am evaluating the maximal data rate that a FIWARE setup could support coming from a single source and how accurate/reliable this communication flow is. My experiment uses the data flow shown below for this specific scenario. This experiment runs on a Linux virtual machine and the containers run via docker-compose.

Custom MQTT Publisher (python script on the host) --> Mosquitto (container) --> JSON IoT Agent (container) --> Orion broker (container) --> QuantumLeap (container) --> Crate DB (container)

I notice that not all data points that are transmitted arrive at Crate DB. After evaluating the logs that are generated by all these components it seems that the JSON IoT Agent is dropping or missing some data points. The MQTT publisher publishes data using a QoS level of 2. Both the publisher and Mosquitto report that all data has been transmitted. However, the IoT Agent only reports on a subset of these data points that have been successfully updated in Orion. It also generates no warning or error when the data points are being transmitted.

During my evaluation, I noticed that the QoS level of the subscription that is used by the JSON IoT Agent is not configured and defaults to 0. I think this might cause the JSON IoT Agent to miss some values when it is being flooded by data.

I would therefore propose to apply the IOTA_MQTT_QOS configuration also to the MQTT subscription. Though, I have not yet been able to verify if this would solve this issue. What do you think?

I have attached the logs of the experiment.
db-crate.log
fiware-quantumleap.log
iotagent-json.log
mqtt-broker.log
orion.log
orion-mongo.log

[fgalan]

Thanks for the report!

So you mean that IOTA_MQTT_QOS is not currently being used to create the MQTT subscription in the MQTT broker? In that case, for what is being used IOTA_MQTT_QOS right now?

[AlvaroVega]

Which mosquitto version are you using ?

[joriwinderickx-fm]

By searching the code on GitHub, I found only two references to the MQTT QoS option ("options.qos") in lib/bindings/MQTTBinding.js (e.g. line 169), where it is used in the function sendConfigurationToDevice(apiKey, deviceId, results, callback) function. However, I am not an expert in javascript.

It would mean that only the downstream communication channel that pushes data towards the devices is using the IOTA_MQTT_QOS setting?

I was using the latest mosquitto container image available in the docker repository (eclipse-mosquitto:latest). It seems to be version 1.6.12 at that time.

[AlvaroVega]

If I'm not wrong, IOTA_MQTT_QOS is used when iotagent sends a message, that is when send a command or a configuration retrieval. When iotagent receives a mqtt message is publisher who should provide a proper qos option.

[joriwinderickx-fm]

While you would be correct if the JSON IoT agent's interface would be programmed as a "server" like in the case of the HTTP interface, it is not the correct in this case. MQTT and also AMQP are publish-subscribe protocols, and the JSON IoT agent only acts as an MQTT client. All messages published by the IoT device will be directed towards the broker and not the IoT agent. Furthermore, the IoT agent sets up a subscription at the MQTT broker to retrieve these published messages and it is on the IoT agent to configure this subscription (QoS level, etc.). The broker will then take care of the distribution of the MQTT messages and act according to the subscription's configuration.

Thank you both for looking into this issue!

[fgalan]

So you mean that is not enough with the IoT Device setting QoS parameter (as @AlvaroVega suggest) but also the subscription done by IOT Agent has to use QoS parameter. Is my understanding correct?

[joriwinderickx-fm]

Yes, that is correct and how I perceive it.

[AlvaroVega]

Then a proper mqtt qos should be used at https://github.com/telefonicaid/iotagent-json/blob/master/lib/bindings/MQTTBinding.js#L140

[fgalan]

So, probably we can do the following:

• Create a new configuration parameter to have control on which QoS setting is used for MQTT subs done by the IOTA. Suggested name: mqtt.qosSub (env var MQTT_QOS_SUB). Modify documentation accordingly (docs/installationguide.md I guess)
• Modify the mqttClient.subscribe() call cited by @AlvaroVega above, using this option instead of null
• Do this change not only in IOTA-JSON but also in IOTA-UL. Both IOTAs should evolve at the same pace.

What do you think @joriwinderickx-fm ? Could you do a pull request contributing with that implementation, please? It seems to be easy... :)