---
title: Mosquitto Integration for Elastic
description: Eclipse Mosquitto is an open source MQTT broker used to move telemetry and command messages between devices, gateways, and applications. The Mosquitto...
url: https://www.elastic.co/docs/reference/integrations/mosquitto
products:
- Elastic integrations
applies_to:
- Serverless Observability projects: Preview
- Serverless Security projects: Preview
- Elastic Stack: Preview since 9.1
---
# Mosquitto Integration for Elastic
| | |
|-------------------------------------------------------------------------------------------------|-------------------------------------------------------------------|
| **Version** | 0.1.0 : Preview ([View all](#changelog)) |
| **Subscription level**[What's this?](https://www.elastic.co/subscriptions) | Basic |
| **Developed by**[What's this?](https://www.elastic.co/docs/reference/integrations/developed-by) | Community |
| **Ingestion method(s)** | File |
| **Minimum Kibana version(s)** | 9.1.08.19.0 |
To use pre-release integrations, go to the **Integrations** page in Kibana, scroll down, and toggle on the _Display beta integrations_ option.
## Overview
[Eclipse Mosquitto](https://mosquitto.org/) is an open source MQTT broker used to move telemetry and command messages between devices, gateways, and applications. The Mosquitto integration for Elastic collects the broker's own log file and parses it into ECS fields, so that client connections, subscriptions, published messages, and broker errors become searchable and can be charted or alerted on.
### Compatibility
This integration has been tested against Mosquitto 2.0.x. It also parses the log wording used by Mosquitto 1.6, including the older disconnect lines that omit the client address.
### How it works
Mosquitto writes its log to the destination named by `log_dest`. The Elastic Agent tails that file with the `filestream` input and forwards each line to Elasticsearch, where the integration's ingest pipeline splits off the timestamp prefix, parses the message body and maps the result onto ECS fields. No connection is made to the broker itself, so the integration works with any listener or authentication configuration.
### Timestamp formats, and why ISO is easier to read
Mosquitto can prefix log lines with either of two timestamp formats, and **this integration parses both** — no configuration change is required to use it.
By default `log_timestamp_format` is unset, and Mosquitto writes the number of seconds since the Unix epoch:
```
1756052767: New connection from 192.0.2.24:43985 on port 1883.
```
Setting `log_timestamp_format` makes the broker write a `strftime`-formatted timestamp instead. The format documented by Mosquitto produces an ISO 8601 style timestamp:
```
log_timestamp_format %Y-%m-%dT%H:%M:%S
```
```
2026-08-24T16:26:07: New connection from 192.0.2.24:43985 on port 1883.
```
Configuring the ISO format is recommended. The resulting log is far easier to read directly on the broker host — epoch seconds must be converted value by value to reveal when each event occurred. Both formats produce exactly the same fields in Elasticsearch, so the choice only affects readability on the broker itself.
One caveat applies to the formatted variant: Mosquitto renders it in the broker host's local time and writes no UTC offset, whereas the epoch default is always UTC. If Elastic Agent runs in a different timezone than the broker, set the **Timezone Offset** option on the integration policy to the broker's timezone so timestamps are interpreted correctly. Adding `%z` to the format, for example `%Y-%m-%dT%H:%M:%S%z`, makes the broker emit an explicit offset, which removes the ambiguity entirely.
## What data does this integration collect?
The Mosquitto integration collects the broker log and parses the following kinds of message:
- Connection lifecycle: new connections, clients and bridges connecting, disconnects with the broker's stated reason, expired and rejected clients.
- MQTT control packets sent and received by the broker, including `PUBLISH`, `CONNACK`, `SUBSCRIBE`, `SUBACK`, `PUBACK`, `PUBREC`, `PUBREL`, `PUBCOMP`, `PINGREQ`, `PINGRESP`, `AUTH` and `DISCONNECT`, together with their topic, QoS, message ID, payload size and flags.
- Subscription detail lines listing each topic filter and the QoS granted or denied.
- Broker lifecycle and configuration: version messages on start-up and shutdown, the configuration file that was loaded, listeners being opened, and records restored from the persistence database.
- Errors, warnings, refused connections and protocol errors.
Lines that do not match a known message shape are still indexed, with the full text kept in `message`.
`log_type subscribe` lines are recognized only when the client identifier contains no spaces, because Mosquitto writes them as ` ` with no delimiter; lines from clients whose identifier contains a space are kept verbatim in `message`.
One known limitation: the line Mosquitto writes for `log_type unsubscribe` consists of nothing but a client identifier and a topic filter separated by a space. That shape cannot be told apart from ordinary two-word log messages, so those lines are indexed with their text in `message` but without a parsed topic. The `UNSUBSCRIBE` packet line that accompanies them is parsed normally.
### Supported use cases
Track which clients connect to and disconnect from an MQTT broker and why, spot clients that reconnect in a loop or are rejected by an access control list, follow the topics being published and subscribed to, and alert on broker errors or restarts. Because source addresses and usernames are mapped to `source.ip`, `user.name` and the `related.*` fields, broker activity can be correlated with the rest of the data in the cluster.
## What do I need to use this integration?
Elastic Agent must be able to read the Mosquitto log file. If the broker runs in a container with `log_dest stdout`, point the integration at the container log files and enable the **Parse Container Logs** option.
## How do I deploy this integration?
### Agent-based deployment
Elastic Agent must be installed. For more details, check the Elastic Agent [installation instructions](https://www.elastic.co/guide/en/fleet/current/elastic-agent-installation.html). You can install only one Elastic Agent per host.
Elastic Agent is required to stream data from the log file receiver and ship the data to Elastic, where the events will then be processed via the integration's ingest pipelines.
### Set up steps in Mosquitto
1. Configure the broker to write its log to a file that Elastic Agent can read, by setting `log_dest` in `mosquitto.conf`:
```
log_dest file /var/log/mosquitto/mosquitto.log
```
When the broker runs in a container, keep `log_dest stdout` and collect the container log file instead.
2. Optionally switch to the more readable ISO 8601 timestamp format:
```
log_timestamp_format %Y-%m-%dT%H:%M:%S
```
3. Choose which message types the broker should log with `log_type`. The default of `error`, `warning`, `notice` and `information` covers connections, disconnects and broker lifecycle messages. Add `subscribe` to record which topic filters the broker grants, or set `log_type all` to include the per-packet `debug` messages. Be aware that `all` is very verbose on a busy broker.
4. Restart Mosquitto and confirm the log file is being written.
#### Vendor resources
- [mosquitto.conf manual page](https://mosquitto.org/man/mosquitto-conf-5.html)
- [Mosquitto documentation](https://mosquitto.org/documentation/)
### Set up steps in Kibana
1. In Kibana, go to **Management** → **Integrations**, search for **Mosquitto** and select **Add Mosquitto**.
2. Set **Paths** to the location of the broker log, for example `/var/log/mosquitto/mosquitto.log`. Container deployments typically use `/var/log/containers/*mosquitto*.log`.
3. Enable **Parse Container Logs** when the paths point at container log files, so the CRI or Docker JSON envelope is removed before parsing.
4. Set **Timezone Offset** if the broker uses `log_timestamp_format` and runs in a different timezone than Elastic Agent. Both a canonical ID such as `Europe/Amsterdam` and an offset such as `-05:00` are accepted.
5. Optionally enable **Preserve original event** to keep the raw log line in `event.original`.
6. Select the agent policy to add the integration to, and save.
### Validation
1. Publish or subscribe to a topic on the broker so that it writes new log lines, or restart the broker.
2. In Kibana, open **Discover** and select the `logs-*` data view.
3. Filter on `event.dataset : "mosquitto.log"` and confirm that documents arrive with a parsed `@timestamp`, an `event.action` and a populated `message`.
## Troubleshooting
- No data is being collected: Confirm that `log_dest` writes to a file and that the user running Elastic Agent has read access to it and to its parent directory. A broker configured with only `log_dest stdout` writes nothing to disk outside a container runtime.
- Timestamps are off by a whole number of hours: The broker is using `log_timestamp_format`, which writes local time with no offset. Set the **Timezone Offset** option to the broker's timezone, or append `%z` to the broker's `log_timestamp_format`.
- Documents are tagged `mosquitto_missing_timestamp`: The broker runs with `log_timestamp false`, or with a `log_timestamp_format` that omits the date. `@timestamp` falls back to the time the line was collected. Configure a format that includes a full date to resolve it.
- Every line arrives wrapped in a JSON envelope: The paths point at container log files. Enable the **Parse Container Logs** option.
- Fewer message types appear than expected: The broker's `log_type` setting controls which messages are written at all. Per-packet `PUBLISH` and `SUBSCRIBE` lines require `log_type all` or the `subscribe` and `unsubscribe` types.
## Performance and scaling
A broker logging with `log_type all` writes several lines for every message it handles, so log volume grows with message throughput rather than with client count. On busy brokers, prefer the default `log_type` and enable `subscribe` only while investigating, or use the **Exclude Lines** capability of the agent policy to drop the highest-volume packet lines.
For more information on architectures that can be used for scaling this integration, check the [Ingest Architectures](https://www.elastic.co/docs/manage-data/ingest/ingest-reference-architectures) documentation.
## Reference
### Inputs used
These inputs can be used with this integration:
## SetupFor more details about the Filestream input settings, check the [Filebeat documentation](https://www.elastic.co/docs/reference/beats/filebeat/filebeat-input-filestream).
### Collecting logs from FilestreamTo collect logs via Filestream, select **Collect logs via Filestream** and configure the following parameters:
- Filestream paths: The full path to the related log file.
### Vendor documentation links
- [Eclipse Mosquitto](https://mosquitto.org/)
- [mosquitto.conf manual page](https://mosquitto.org/man/mosquitto-conf-5.html)
- [Mosquitto logging documentation](https://mosquitto.org/man/mosquitto-conf-5.html)
### Data streams
#### log
The `log` data stream provides events from the Eclipse Mosquitto broker log, covering connection lifecycle, MQTT control packets, subscriptions, broker lifecycle and errors.
Mosquitto's file and standard output destinations write no severity marker, so `log.level` is derived from the wording of each message: lines beginning with `Error` or `Connection Refused:` become `error`, lines beginning with `Warning` or `Protocol error from` become `warning`, and everything else becomes `information`. Use `log_dest syslog` if the broker's own severity classification is required.
##### log fields
| Field | Description | Type | Unit |
|--------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------|------|
| @timestamp | Date/time when the event originated. This is the date/time extracted from the event, typically representing when the event was generated by the source. If the event source has no original timestamp, this value is typically populated by the first time the event was received by the pipeline. Required field for all events. | date | |
| data_stream.dataset | The field can contain anything that makes sense to signify the source of the data. Examples include `nginx.access`, `prometheus`, `endpoint` etc. For data streams that otherwise fit, but that do not have dataset set we use the value "generic" for the dataset value. `event.dataset` should have the same value as `data_stream.dataset`. Beyond the Elasticsearch data stream naming criteria noted above, the `dataset` value has additional restrictions: * Must not contain `-` * No longer than 100 characters | constant_keyword | |
| data_stream.namespace | A user defined namespace. Namespaces are useful to allow grouping of data. Many users already organize their indices this way, and the data stream naming scheme now provides this best practice as a default. Many users will populate this field with `default`. If no value is used, it falls back to `default`. Beyond the Elasticsearch index naming criteria noted above, `namespace` value has the additional restrictions: * Must not contain `-` * No longer than 100 characters | constant_keyword | |
| data_stream.type | An overarching type for the data stream. Currently allowed values are "logs" and "metrics". We expect to also add "traces" and "synthetics" in the near future. | constant_keyword | |
| ecs.version | ECS version this event conforms to. `ecs.version` is a required field and must exist in all events. When querying across multiple indices -- which may conform to slightly different ECS versions -- this field lets integrations adjust to the schema version of the events. | keyword | |
| error.message | Error message. | match_only_text | |
| event.action | The action captured by the event. This describes the information in the event. It is more specific than `event.category`. Examples are `group-add`, `process-started`, `file-created`. The value is normally defined by the implementer. | keyword | |
| event.category | This is one of four ECS Categorization Fields, and indicates the second level in the ECS category hierarchy. `event.category` represents the "big buckets" of ECS categories. For example, filtering on `event.category:process` yields all events relating to process activity. This field is closely related to `event.type`, which is used as a subcategory. This field is an array. This will allow proper categorization of some events that fall in multiple categories. | keyword | |
| event.dataset | Name of the dataset. If an event source publishes more than one type of log or events (e.g. access log, error log), the dataset is used to specify which one the event comes from. It's recommended but not required to start the dataset name with the module name, followed by a dot, then the dataset name. | constant_keyword | |
| event.kind | This is one of four ECS Categorization Fields, and indicates the highest level in the ECS category hierarchy. `event.kind` gives high-level information about what type of information the event contains, without being specific to the contents of the event. For example, values of this field distinguish alert events from metric events. The value of this field can be used to inform how these kinds of events should be handled. They may warrant different retention, different access control, it may also help understand whether the data is coming in at a regular interval or not. | keyword | |
| event.module | Name of the module this data is coming from. If your monitoring agent supports the concept of modules or plugins to process events of a given source (e.g. Apache logs), `event.module` should contain the name of this module. | constant_keyword | |
| event.original | Raw text message of entire event. Used to demonstrate log integrity or where the full log message (before splitting it up in multiple parts) may be required, e.g. for reindex. This field is not indexed and doc_values are disabled. It cannot be searched, but it can be retrieved from `_source`. If users wish to override this and index this field, please see `Field data types` in the `Elasticsearch Reference`. | keyword | |
| event.outcome | This is one of four ECS Categorization Fields, and indicates the lowest level in the ECS category hierarchy. `event.outcome` simply denotes whether the event represents a success or a failure from the perspective of the entity that produced the event. Note that when a single transaction is described in multiple events, each event may populate different values of `event.outcome`, according to their perspective. Also note that in the case of a compound event (a single event that contains multiple logical events), this field should be populated with the value that best captures the overall success or failure from the perspective of the event producer. Further note that not all events will have an associated outcome. For example, this field is generally not populated for metric events, events with `event.type:info`, or any events for which an outcome does not make logical sense. | keyword | |
| event.timezone | This field should be populated when the event's timestamp does not include timezone information already (e.g. default Syslog timestamps). It's optional otherwise. Acceptable timezone formats are: a canonical ID (e.g. "Europe/Amsterdam"), abbreviated (e.g. "EST") or an HH:mm differential (e.g. "-05:00"). | keyword | |
| event.type | This is one of four ECS Categorization Fields, and indicates the third level in the ECS category hierarchy. `event.type` represents a categorization "sub-bucket" that, when used along with the `event.category` field values, enables filtering events down to a level appropriate for single visualization. This field is an array. This will allow proper categorization of some events that fall in multiple event types. | keyword | |
| input.type | Type of filebeat input. | keyword | |
| log.file.device_id | Device Id of the log file this event came from. | keyword | |
| log.file.fingerprint | The sha256 fingerprint identity of the file when fingerprinting is enabled. | keyword | |
| log.file.inode | Inode number of the log file. | keyword | |
| log.file.path | Full path to the log file this event came from, including the file name. It should include the drive letter, when appropriate. If the event wasn't read from a log file, do not populate this field. | keyword | |
| log.level | Original log level of the log event. If the source of the event provides a log level or textual severity, this is the one that goes in `log.level`. If your source doesn't specify one, you may put your event transport's severity here (e.g. Syslog severity). Some examples are `warn`, `err`, `i`, `informational`. | keyword | |
| log.offset | Log offset. | long | |
| message | For log events the message field contains the log message, optimized for viewing in a log viewer. For structured logs without an original message field, other fields can be concatenated to form a human-readable summary of the event. If multiple messages exist, they can be combined into one message. | match_only_text | |
| mosquitto.log.auth_method | Authentication method named in an MQTT v5 AUTH packet. | keyword | |
| mosquitto.log.clean_session | Whether the client requested a clean session, so the broker discards any previously stored session state. | boolean | |
| mosquitto.log.client_id | MQTT client identifier the broker used for this client, as supplied in its CONNECT packet. | keyword | |
| mosquitto.log.config_file | Path to the configuration file the broker loaded. | keyword | |
| mosquitto.log.dup | Whether the MQTT DUP flag was set, marking a redelivered message. | boolean | |
| mosquitto.log.keepalive | Keepalive interval negotiated with the client, in seconds. | long | s |
| mosquitto.log.message_id | MQTT packet identifier, logged by Mosquitto as `m\` or `Mid: \`. | long | |
| mosquitto.log.packet_type | MQTT control packet the broker sent or received, for example `PUBLISH`, `CONNACK` or `SUBSCRIBE`. | keyword | |
| mosquitto.log.payload_size | Size of the MQTT message payload in bytes. | long | byte |
| mosquitto.log.persistence_file | Path the broker wrote its in-memory database to. | keyword | |
| mosquitto.log.protocol_version | Raw MQTT protocol level the client connected with, as logged by Mosquitto in the `p\` field. | long | |
| mosquitto.log.qos | MQTT quality of service level (`0`, `1` or `2`). On subscription lines this is the QoS granted by the broker. | long | |
| mosquitto.log.reason | Human readable reason the broker gave for a disconnect, a refused connection or a protocol error. | keyword | |
| mosquitto.log.reason_code | MQTT reason code carried by the packet. The meaning depends on the packet type and on the MQTT version in use. | long | |
| mosquitto.log.restored_count | Number of records restored from the persistence database on start-up. | long | |
| mosquitto.log.restored_type | Kind of records the restore count refers to, for example `clients` or `retained messages`. | keyword | |
| mosquitto.log.retain | Whether the MQTT retain flag was set on the message. | boolean | |
| mosquitto.log.session_present | Whether the broker resumed an existing session for the client, taken from the first field of the CONNACK log line. | boolean | |
| mosquitto.log.socket_path | Filesystem path of the Unix domain socket the broker is listening on. | keyword | |
| mosquitto.log.topic | MQTT topic name or topic filter the log line refers to. | keyword | |
| mosquitto.log.version | Mosquitto broker version reported on start-up and shutdown. | keyword | |
| network.direction | Direction of the network traffic. When mapping events from a host-based monitoring context, populate this field from the host's point of view, using the values "ingress" or "egress". When mapping events from a network or perimeter-based monitoring context, populate this field from the point of view of the network perimeter, using the values "inbound", "outbound", "internal" or "external". Note that "internal" is not crossing perimeter boundaries, and is meant to describe communication between two hosts within the perimeter. Note also that "external" is meant to describe traffic between two hosts that are external to the perimeter. This could for example be useful for ISPs or VPN service providers. | keyword | |
| related.ip | All of the IPs seen on your event. | ip | |
| related.user | All the user names or other user identifiers seen on the event. | keyword | |
| server.port | Port of the server. | long | |
| source.ip | IP address of the source (IPv4 or IPv6). | ip | |
| source.port | Port of the source. | long | |
| tags | List of keywords used to tag each event. | keyword | |
| user.name | Short name or login of the user. | keyword | |
| user.name.text | Multi-field of `user.name`. | match_only_text | |
##### log sample event
```json
{
"@timestamp": "2026-08-24T16:26:07.000Z",
"agent": {
"ephemeral_id": "8e55e77a-968d-4b4a-9766-cfe24328911b",
"id": "7ee5c62d-fa5e-4536-abb1-bfe1398174e3",
"name": "elastic-agent-33099",
"type": "filebeat",
"version": "8.19.0"
},
"data_stream": {
"dataset": "mosquitto.log",
"namespace": "37129",
"type": "logs"
},
"ecs": {
"version": "9.3.0"
},
"elastic_agent": {
"id": "7ee5c62d-fa5e-4536-abb1-bfe1398174e3",
"snapshot": false,
"version": "8.19.0"
},
"event": {
"action": "broker-starting",
"agent_id_status": "verified",
"category": [
"process"
],
"dataset": "mosquitto.log",
"ingested": "2026-11-03T09:25:27Z",
"kind": "event",
"original": "2026-08-24T16:26:07: mosquitto version 2.0.22 starting",
"timezone": "UTC",
"type": [
"start"
]
},
"host": {
"architecture": "aarch64",
"containerized": false,
"hostname": "elastic-agent-33099",
"ip": [
"10.89.2.2",
"fe80::68aa:49ff:fe4c:80af",
"10.89.1.13",
"fe80::a473:e6ff:fe39:ee27"
],
"mac": [
"6A-AA-49-4C-80-AF",
"A6-73-E6-39-EE-27"
],
"name": "elastic-agent-33099",
"os": {
"family": "",
"kernel": "6.17.7-300.fc43.aarch64",
"name": "Wolfi",
"platform": "wolfi",
"type": "linux",
"version": "20230201"
}
},
"input": {
"type": "filestream"
},
"log": {
"file": {
"device_id": "39",
"inode": "4385",
"path": "/tmp/service_logs/mosquitto-iso.log"
},
"level": "information",
"offset": 0
},
"message": "mosquitto version 2.0.22 starting",
"mosquitto": {
"log": {
"version": "2.0.22"
}
},
"tags": [
"preserve_original_event",
"mosquitto-log"
]
}
```
## Changelog
| Version | Details | Minimum Kibana version |
|-----------|-----------------------------------------------------------------------------------------------------------|------------------------|
| **0.1.0** | **Enhancement** ([View pull request](https://github.com/elastic/integrations/pull/20937))Initial release. | 9.1.08.19.0 |