MQTT

MQTT Message Queue

Synopsis

Creates a target that publishes log messages to MQTT topics with support for batching, QoS levels, TLS encryption, and automatic retry logic.

Schema

- name: <string>
  description: <string>
  type: mqtt
  pipelines: <pipeline[]>
  status: <boolean>
  properties:
    url: <string>
    topic: <string>
    client_id: <string>
    username: <string>
    password: <string>
    qos: <integer>
    retained: <boolean>
    clean_session: <boolean>
    auto_reconnect: <boolean>
    keep_alive: <integer>
    timeout: <integer>
    batch_size: <integer>
    max_retries: <integer>
    retry_delay: <integer>
    field_format: <string>
    tls:
      status: <boolean>
      verify: <boolean>
      cert_name: <string>
      key_name: <string>
      min_tls_version: <string>
      max_tls_version: <string>
    debug:
      status: <boolean>
      dont_send_logs: <boolean>

Configuration

The following fields are used to define the target:

Field	Required	Default	Description
name	Y		Target name
description	N	-	Optional description
type	Y		Must be mqtt
pipelines	N	-	Optional post-processor pipelines
status	N	true	Enable/disable the target

Connection

Field	Required	Default	Description
url	Y	-	MQTT broker URL (e.g., tcp://localhost:1883, ssl://broker:8883, ws://broker:9001)
topic	Y	-	MQTT topic for message publishing
client_id	N	Auto-generated	MQTT client identifier (auto-generated if not specified)
username	N	-	MQTT username for authentication
password	N	-	MQTT password for authentication
timeout	N	30	Connection and write timeout in seconds
keep_alive	N	60	Keep-alive interval in seconds

MQTT Settings

Field	Required	Default	Description
qos	N	1	Quality of Service level: 0 (at most once), 1 (at least once), 2 (exactly once)
retained	N	false	Retain messages on broker for new subscribers
clean_session	N	true	Start with clean session (discard previous session state)
auto_reconnect	N	true	Enable automatic reconnection on connection loss

Batch Configuration

Field	Required	Default	Description
batch_size	N	1000	Number of messages to batch before publishing (minimum 1)
max_retries	N	3	Maximum retry attempts for failed publish operations
retry_delay	N	1	Delay between retry attempts in seconds

Processing

Field	Required	Default	Description
field_format	N	-	Data normalization format. See applicable Normalization section

TLS Configuration

Field	Required	Default	Description
tls.status	N	false	Enable TLS encryption

TLS material is resolved through the shared client builder, so these keys mean the same thing on every target that uses it. They are nested under a tls: block.

Field	Required	Default	Description
tls.verify	N	true	Verify the server certificate. On by default — set it to false only to accept an untrusted certificate, and only where you control the network path
tls.server_name	N	-	SNI hostname override for the handshake. Use it when the certificate's name does not match the address you connect to
tls.ca_name	N	-	CA bundle used to verify the server certificate. When unset the host trust store is used; when set it replaces the host trust store rather than adding to it
tls.cert_name	N*	-	Client certificate, for mutual TLS
tls.key_name	N*	-	Client private key, for mutual TLS
tls.passphrase	N	-	Passphrase for an encrypted client private key
tls.min_tls_version	N	tls1.2	Lowest protocol version accepted (tls1.0, tls1.1, tls1.2, tls1.3)
tls.max_tls_version	N	tls1.3	Highest protocol version accepted (tls1.0, tls1.1, tls1.2, tls1.3)

* cert_name and key_name are individually optional but must be supplied together — a certificate without its key, or a key without its certificate, is a configuration error.

Scheduling

See Scheduling and Pool Behavior for interval and cron fields shared by all targets.

Debug Options

Field	Required	Default	Description
debug.status	N	false	Enable debug logging
debug.dont_send_logs	N	false	Process logs but don't send to target (testing)

Details

The MQTT target publishes log messages to MQTT topics using the Eclipse Paho MQTT client library. Messages are accumulated in batches and published when the batch size is reached. Each message is published with automatic retry logic and configurable Quality of Service levels.

The target maintains a persistent connection to the MQTT broker with automatic reconnection capabilities. Connection options include configurable timeouts, keep-alive intervals, and session management.

Quality of Service Levels

• QoS 0 (At most once): Fire-and-forget, no acknowledgment, lowest overhead
• QoS 1 (At least once): Acknowledged delivery, message may be duplicated
• QoS 2 (Exactly once): Guaranteed single delivery, highest overhead

Prerequisites

1. A running MQTT broker (e.g., Mosquitto, HiveMQ, AWS IoT Core)
2. Network connectivity to the MQTT broker
3. Valid authentication credentials if the broker requires authentication
4. TLS certificates if using encrypted connections

note

The client_id is auto-generated if not specified. For persistent sessions (clean_session: false), use a consistent client ID across reconnections.

warning

Both tls.cert_name and tls.key_name must be provided together when using client certificate authentication. Providing only one will result in a configuration error.

note

The retained flag causes the broker to store the last message on the topic, delivering it to new subscribers immediately. Use with caution for high-frequency log data.

Examples

Basic

Minimum configuration for publishing to MQTT:

targets:
  - name: basic_mqtt
    type: mqtt
    properties:
      url: "tcp://localhost:1883"
      topic: "logs/system"

With Authentication

Configuration with username/password authentication:

targets:
  - name: authenticated_mqtt
    type: mqtt
    properties:
      url: "tcp://mqtt.example.com:1883"
      topic: "logs/application"
      username: "logger"
      password: "secure_password"

With TLS

Configuration with TLS encryption:

targets:
  - name: secure_mqtt
    type: mqtt
    properties:
      url: "ssl://mqtt.example.com:8883"
      topic: "logs/secure"
      username: "logger"
      password: "secure_password"
      tls:
        status: true
        verify: true
        min_tls_version: "tls1.2"

With Client Certificate

Configuration using mutual TLS with client certificates:

targets:
  - name: mtls_mqtt
    type: mqtt
    properties:
      url: "ssl://mqtt.example.com:8883"
      topic: "logs/secure"
      tls:
        status: true
        verify: true
        cert_name: "client.pem"
        key_name: "client-key.pem"
        min_tls_version: "tls1.2"

High Reliability

Configuration with QoS 2 for guaranteed delivery:

targets:
  - name: reliable_mqtt
    type: mqtt
    properties:
      url: "tcp://mqtt.example.com:1883"
      topic: "logs/critical"
      qos: 2
      clean_session: false
      client_id: "datastream-logger-001"

Retained Messages

Configuration with retained messages for topic state:

targets:
  - name: retained_mqtt
    type: mqtt
    properties:
      url: "tcp://mqtt.example.com:1883"
      topic: "logs/status"
      retained: true
      qos: 1

Custom Batch Size

Configuration with custom batching and retry settings:

targets:
  - name: batch_mqtt
    type: mqtt
    properties:
      url: "tcp://mqtt.example.com:1883"
      topic: "logs/highvolume"
      batch_size: 5000
      max_retries: 5
      retry_delay: 2
      timeout: 60

With Normalization

Configuration using field normalization:

targets:
  - name: normalized_mqtt
    type: mqtt
    properties:
      url: "tcp://mqtt.example.com:1883"
      topic: "logs/ecs"
      field_format: "ecs"

WebSocket Connection

Configuration using WebSocket transport:

targets:
  - name: ws_mqtt
    type: mqtt
    properties:
      url: "ws://mqtt.example.com:9001"
      topic: "logs/websocket"
      username: "logger"
      password: "secure_password"

With Pipeline

Using a pipeline for additional log processing:

targets:
  - name: pipeline_mqtt
    type: mqtt
    pipelines:
      - enrich_logs
    properties:
      url: "tcp://mqtt.example.com:1883"
      topic: "logs/enriched"