Monitoring and logging MQTT topic messages using Eclipse Mosquitto

Data required

Cisco Meraki camera data, or any other camera device

Key terminology

• MQTT: This lightweight messaging protocol is commonly used in IoT environments to allow devices to communicate. MQTT messages are structured as "topics" that devices subscribe to or publish on.
• Eclipse Mosquitto: An open-source message broker that implements MQTT. This article uses the mosquitto_sub command to subscribe to topics on an MQTT broker and format the messages for logging.

Prerequisites

In this scenario we are going to explore adding logging data to message data from an MQTT broker. This is done by modifying the command parameters for mosquitto_sub to add in the required formatting.

To complete this process in your environment, you should be comfortable with using the command line, and command line tools. You'll also need:

• Command line access. This is needed to subscribe to a topic on an existing MQTT broker. If you do not have access to the internet, you might need a host where an MQTT broker can be installed.
• The ability to subscribe to a MQTT broker. Your subscription could be to a local MQTT broker, a public test server, or any other MQTT broker that you are able to access.
• The ability to subscribe to an MQTT message on a private or a public test server. The examples used in this article are messages from Meraki MV cameras published to a test server.

Logging MQTT messages using Mosquitto and Splunk Enterprise

To enable the logging of messages from MQTT topics, you need to add formatting and metadata. After formatting, the log will be outputted to a log file.

In the following example, mosquitto_sub is used to subscribe to an MQTT broker, format messages in JSON, and log them. By formatting these messages in JSON, users can ingest, search, and visualize data more easily in Splunk Enterprise dashboards, making it accessible to monitor IoT devices like Meraki MV cameras.

The log file is then ingested into Splunk Enterprise, where it is queried and visualized to build an example dashboard with a filter.

1. Subscribe to a wildcard topic on your device's MQTT broker host, or use test data from test.mosquitto.org. This example uses a public test server. The formatting adds a timestamp and outputs to a log file. JSON formatting, applied via mosquitto_sub’s output formatting, simplifies import into Splunk Enterprise. The example below uses a Unix timestamp %U and the MQTT message payload %j. Additional parameters and metadata can be included to provide context for your services, applications, asset location, or other valuable information.

   mosquitto_sub -h test.mosquitto.org -v -F '{"Timestamp": %U ,"Attributes":{"key":"value"},"Resource":{"service.name":"mosquitto_sub_topic_to_log","service.version":"semver:0.0.1"},"SeverityText":"INFO","SeverityNumber":9,"Body": %j }' -t '#' >> /path/to/mosquitto_mqtt_topics_to_log.json
   

2. Logs should now be written when subscribed topics publish messages. To confirm this, check for log messages in the log file from mosquitto_sub. You can view the last few entries using the tail command:

   tail /path/to/mosquitto_mqtt_topics_to_log.json
   

   The screenshot below shows an example of a Zsh terminal tailing a JSON-formatted log file from Eclipse Mosquitto.

   

3. After verifying that the log file contains data, import it into Splunk Enterprise from a local file directory. To do this, go to Settings > Data Inputs > Files & Directories > Add New +. The mosquitto_sub log file can also be sent to Splunk Enterprise using the Splunk Universal Forwarder or Splunk OpenTelemetry Collector.

   The screenshot below shows an example of log file names for MQTT message logs intended for ingestion into Splunk Enterprise.

   

4. The JSON body, after it has been ingested by Splunk Enterprise, will appear like this in the search events:

   {
     Attributes: { key: value },
     Body: { payload: {"lux": 3.5}, payloadlen: 12, qos: 0, retain: 0, topic: /merakimv/QOE0-89RF-KMHD/light, tst: 2024-10-07T08:19:03.788382-0700 },
     Resource: { service.name: mosquitto_sub_wildcard_topic_to_log, service.version: semver:0.0.1 },
     SeverityNumber: 9,
     SeverityText: INFO,
     Timestamp: 1728314343.788357
   }         
   

5. If you know the topic of a device or sensor, you can query it in Splunk Enterprise by the topic included with each pre-formatted and logged MQTT message. For example, messages from a Meraki MV camera include MQTT topics for audio, light, and people detection. After importing data into Splunk Enterprise, you can query the sensor readings over time with a SPL query.

   The example SPL below shows a search under the MQTT topic "/merakimv/*/light". This topic contains a device identifier for the Meraki MV camera. The device ID can be extracted into a new field titled device_id using the rex command. The MQTT message payload is formatted in an encoded JSON format, so to utilize the payload data in a query, we will need to decode it. The fromjson command allows us to extract the encoded JSON payload and utilize the key values:

   index="your_mqtt_index" host="your_mqtt_sub_host" sourcetype="mqtt_json" "Body.topic"="/merakimv/*/light"
   | rex field=Body.topic "(?<device_id>[0-9A-Z]{4}-[0-9A-Z]{4}-[0-9A-Z]{4})"
   | fromjson Body.payload
   | table Body.tst, device_id, lux
   

   The screenshot below shows an example of MQTT message logs for Meraki MV camera lux readings ingested into Splunk Enterprise.

   

6. A query for Meraki MV audio detections might look similar to the one shown in the previous step, changing the MQTT topic to "/merakimv/*/audio_detections". This returns a payload with timestamp, detection ID, confidence score, and detection type. The query should look like the following:

   index="your_mqtt_index" host="your_mqtt_sub_host" sourcetype="mqtt_json" "Body.topic"="/merakimv/*/audio_detections"
   | rex field=Body.topic "(?<device_id>[0-9A-Z]{4}-[0-9A-Z]{4}-[0-9A-Z]{4})"
   | rex mode=sed field=Body.payload "s/(\[|\])//"
   | fromjson Body.payload
   | table ts, device_id, id, confidence, name
   

   The screenshot below shows an example of MQTT message logs from Meraki MV audio detection being ingested into Splunk Enterprise.

   

7. To query Meraki MV audio analytics, use the MQTT topic "/merakimv/*/audio_analytics" for payload data on audio levels at specific times:

   index="your_mqtt_index" host="your_mqtt_sub_host" sourcetype="mqtt_json" "Body.topic"="/merakimv/*/audio_analytics"
   | rex field=Body.topic "(?<device_id>[0-9A-Z]{4}-[0-9A-Z]{4}-[0-9A-Z]{4})"
   | fromjson Body.payload
   | table ts, audioLevel
   | chart max(audioLevel) BY ts
   

   The screenshot below shows an example of MQTT message logs for Meraki MV audio analytics ingested into Splunk Enterprise.

   

8. An SPL query for the Meraki MV raw_detections topic can use a similar structure, changing the topic and decoding the message object. This query excludes payload lengths with less than 42 characters, as these are empty message bodies with a timestamp. The query should look like the following:

   index="your_mqtt_index" host="your_mqtt_sub_host" sourcetype="mqtt_json" "Body.topic"="/merakimv/*/raw_detections" Body.payloadlen>42
   | rex field=Body.topic "(?<device_id>[0-9A-Z]{4}-[0-9A-Z]{4}-[0-9A-Z]{4})"
   | fromjson Body.payload
   | table device_id, ts, detectionId, confidence, detectionType
   

   The screenshot below shows an example of MQTT message logs for Meraki camera audio raw detections ingested into Splunk Enterprise.

   

[Unit]
Description=Mosquitto MQTT Subscriber to topic: # 
After=network.target
Wants=network.target
[Service]
Type=notify
NotifyAccess=main
ExecStart=mosquitto_sub -h your.mqttbrokerhost.com -v -F '{"Timestamp": %U ,"Attributes":{"key":"value"},"Resource":{"service.name":"mosquitto_sub_topic_to_log","service.version":"semver:0.0.1"},"SeverityText":"INFO","SeverityNumber":9,"Body": %j }' -t '#' >> /path/to/LogAllMqttTopics.json
ExecReload=/bin/kill -HUP $MAINPID
Restart=on-failure
[Install]
WantedBy=multi-user.target