MQTT Integration

Subscribe an Arc Enterprise cluster to MQTT brokers through the API: map topics to measurements, extract tags from topic paths, and manage subscriptions without restarting nodes.

Ingest data directly from MQTT brokers into Arc. Connect to IoT devices, industrial sensors, and message brokers without middleware.

Overview

Arc provides native MQTT subscription with dynamic, API-driven configuration. Manage multiple MQTT brokers and subscriptions at runtime without server restarts.

Key features:

API-driven subscription management - Create, update, delete, start/stop subscriptions via REST API
Multiple simultaneous brokers - Connect to different MQTT brokers for different data sources
Topic wildcards - Subscribe using + (single level) and # (multi-level) wildcards
Auto-detection - Automatically detects JSON and MessagePack message formats
High performance - MessagePack columnar format for high-throughput ingestion
Topic mapping - Extract tags from topic path segments
TLS/SSL support - Client certificates and CA verification
Encrypted credentials - Passwords encrypted at rest using AES-256-GCM
Auto-reconnect - Exponential backoff on connection loss
QoS support - QoS 0, 1, and 2

Prerequisites

Arc server running (v26.02.1 or higher)
Arc API token (if authentication is enabled)
MQTT broker accessible from Arc server

Quick start

2. Create a subscription

curl -X POST http://localhost:8000/api/v1/mqtt/subscriptions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{
    "name": "factory-sensors",
    "broker": "tcp://localhost:1883",
    "topics": ["sensors/#"],
    "database": "iot",
    "auto_start": true
  }'

Response:

{
  "id": "sub_abc123",
  "name": "factory-sensors",
  "broker": "tcp://localhost:1883",
  "topics": ["sensors/#"],
  "database": "iot",
  "status": "running",
  "created_at": "2026-02-13T10:00:00Z"
}

3. Send test data

Publish a message to your MQTT broker:

mosquitto_pub -h localhost -t "sensors/temperature" \
  -m '{"time": 1706745600000000, "value": 23.5, "device_id": "sensor-001"}'

4. Query the data

curl -X POST http://localhost:8000/api/v1/query \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "sql": "SELECT * FROM iot.temperature ORDER BY time DESC LIMIT 10",
    "format": "json"
  }'

API reference

Subscription management

Method	Endpoint	Description
`POST`	`/api/v1/mqtt/subscriptions`	Create a new subscription
`GET`	`/api/v1/mqtt/subscriptions`	List all subscriptions
`GET`	`/api/v1/mqtt/subscriptions/{id}`	Get subscription details
`PUT`	`/api/v1/mqtt/subscriptions/{id}`	Update subscription
`DELETE`	`/api/v1/mqtt/subscriptions/{id}`	Delete subscription

Lifecycle control

Method	Endpoint	Description
`POST`	`/api/v1/mqtt/subscriptions/{id}/start`	Start subscription
`POST`	`/api/v1/mqtt/subscriptions/{id}/stop`	Stop subscription
`POST`	`/api/v1/mqtt/subscriptions/{id}/restart`	Restart subscription

Monitoring

Method	Endpoint	Description
`GET`	`/api/v1/mqtt/subscriptions/{id}/stats`	Get subscription stats
`GET`	`/api/v1/mqtt/stats`	Aggregate stats (all subscriptions)
`GET`	`/api/v1/mqtt/health`	Health check

Subscription options

Create subscription request

{
  "name": "factory-sensors",
  "broker": "tcp://localhost:1883",
  "topics": ["sensors/#", "factory/+/metrics"],
  "database": "iot",
  "qos": 1,
  "client_id": "arc-factory",
  "username": "mqtt_user",
  "password": "mqtt_pass",
  "tls_enabled": false,
  "tls_cert_path": "/path/to/client.crt",
  "tls_key_path": "/path/to/client.key",
  "tls_ca_path": "/path/to/ca.crt",
  "topic_mapping": {},
  "keep_alive_seconds": 60,
  "connect_timeout_seconds": 30,
  "reconnect_max_seconds": 60,
  "auto_start": true
}

Field reference

Field	Type	Required	Default	Description
`name`	string	Yes	-	Unique subscription name
`broker`	string	Yes	-	MQTT broker URL (tcp://, ssl://, ws://, wss://)
`topics`	array	Yes	-	List of topics to subscribe
`database`	string	Yes	-	Target Arc database
`qos`	int	No	1	QoS level: 0, 1, or 2
`client_id`	string	No	auto	MQTT client ID
`username`	string	No	-	MQTT username
`password`	string	No	-	MQTT password (encrypted at rest)
`tls_enabled`	bool	No	false	Enable TLS/SSL
`tls_cert_path`	string	No	-	Client certificate path
`tls_key_path`	string	No	-	Client key path
`tls_ca_path`	string	No	-	CA certificate path
`topic_mapping`	object (`{string: string}`)	No	{}	Per-topic target-database override: maps an exact MQTT topic to a database name, overriding `database` for messages on that topic. See Topic Mapping.
`keep_alive_seconds`	int	No	60	MQTT keep-alive interval
`connect_timeout_seconds`	int	No	30	Connection timeout
`reconnect_max_seconds`	int	No	60	Maximum reconnect backoff delay. The reconnect delay starts at 1 second and doubles up to this cap (the 1-second minimum is fixed by the MQTT client library and is not configurable).
`auto_start`	bool	No	true	Start on creation and server restart

Message formats

Arc automatically detects the message format based on content.

JSON single record

{
  "time": 1706745600000000,
  "temperature": 23.5,
  "humidity": 65.2,
  "device_id": "sensor-001"
}

JSON batch

[
  {"time": 1706745600000000, "temperature": 23.5},
  {"time": 1706745601000000, "temperature": 23.6},
  {"time": 1706745602000000, "temperature": 23.4}
]

MessagePack row-based

Same structure as JSON, but MessagePack encoded. Detected via magic bytes.

MessagePack columnar (fastest)

{
  "m": "temperature",
  "columns": {
    "time": [1706745600000000, 1706745601000000],
    "value": [23.5, 23.6],
    "device_id": ["sensor-001", "sensor-001"]
  }
}

Performance: MessagePack columnar format sustains high ingest throughput.

Timestamp handling

If time field is present: used as-is (auto-detects milliseconds/microseconds/nanoseconds)
If time field is missing: current UTC time is used

Measurement, tags, and fields

Arc derives the measurement, tags, and fields from the message payload, not from the topic structure. The topic itself is not parsed for the measurement name or for tag values.

For each decoded message:

Measurement — taken from the payload's m field, or measurement field. If neither is present, it defaults to mqtt.
Tags — taken from a tags object in the payload (string values).
Fields — taken from a fields object if present; otherwise every remaining top-level key (anything other than m/measurement, t/time/timestamp, tags, fields) is treated as a field.
Timestamp — from t, time, or timestamp (auto-detects ms/µs/ns); current UTC time if absent.

So to land in measurement machine_metrics with tags line and machine_id, publish a payload like:

{
  "m": "machine_metrics",
  "time": 1706745600000000,
  "tags": { "line": "A", "machine_id": "42" },
  "fields": { "temperature": 71.5, "rpm": 1480 }
}

A flat payload with no m/tags/fields (e.g. {"time": ..., "temperature": 23.5}) is also accepted: it lands in the default mqtt measurement with the remaining keys as fields and no tags.

Topic mapping (per-topic database override)

topic_mapping maps an exact MQTT topic string to a target database name, overriding the subscription's database for messages received on that topic. It is a flat { "<topic>": "<database>" } object — it does not configure measurements or tags.

{
  "name": "factory-sensors",
  "broker": "tcp://localhost:1883",
  "topics": ["factory/line1/metrics", "factory/line2/metrics"],
  "database": "iot",
  "topic_mapping": {
    "factory/line2/metrics": "iot_line2"
  }
}

In this example, messages on factory/line1/metrics are written to the default iot database, while messages on factory/line2/metrics are routed to iot_line2.

Authentication

Basic authentication

curl -X POST http://localhost:8000/api/v1/mqtt/subscriptions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{
    "name": "authenticated-broker",
    "broker": "tcp://broker.example.com:1883",
    "topics": ["data/#"],
    "database": "production",
    "username": "mqtt_user",
    "password": "mqtt_password"
  }'

Password encryption

Passwords are encrypted at rest using AES-256-GCM. Set the encryption key:

# Generate a 32-byte key
openssl rand -base64 32

# Set environment variable before starting Arc
export ARC_ENCRYPTION_KEY="your-base64-encoded-32-byte-key"

TLS/SSL configuration

Server certificate verification

curl -X POST http://localhost:8000/api/v1/mqtt/subscriptions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{
    "name": "secure-broker",
    "broker": "ssl://broker.example.com:8883",
    "topics": ["secure/#"],
    "database": "production",
    "tls_enabled": true,
    "tls_ca_path": "/etc/arc/certs/ca.crt"
  }'

Client certificate authentication (mTLS)

curl -X POST http://localhost:8000/api/v1/mqtt/subscriptions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{
    "name": "mtls-broker",
    "broker": "ssl://broker.example.com:8883",
    "topics": ["secure/#"],
    "database": "production",
    "tls_enabled": true,
    "tls_cert_path": "/etc/arc/certs/client.crt",
    "tls_key_path": "/etc/arc/certs/client.key",
    "tls_ca_path": "/etc/arc/certs/ca.crt"
  }'

Configuration examples

IoT sensor network

curl -X POST http://localhost:8000/api/v1/mqtt/subscriptions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{
    "name": "iot-sensors",
    "broker": "tcp://mosquitto:1883",
    "topics": [
      "sensors/+/temperature",
      "sensors/+/humidity",
      "sensors/+/pressure"
    ],
    "database": "iot",
    "qos": 1
  }'

Devices set the measurement and tags in the payload — e.g. a temperature sensor publishes:

{ "m": "temperature", "tags": { "sensor_id": "temp-001" }, "fields": { "value": 23.5 } }

Industrial factory

curl -X POST http://localhost:8000/api/v1/mqtt/subscriptions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{
    "name": "factory-floor",
    "broker": "tcp://factory-mqtt:1883",
    "topics": ["factory/+/+/metrics"],
    "database": "manufacturing",
    "qos": 2
  }'

Machines publish the measurement and tags in the payload:

{
  "m": "machine_metrics",
  "tags": { "line": "A", "machine_id": "42" },
  "fields": { "temperature": 71.5, "rpm": 1480 }
}

Monitoring

Subscription stats

# Stats for a specific subscription
curl -H "Authorization: Bearer $TOKEN" \
  http://localhost:8000/api/v1/mqtt/subscriptions/{id}/stats

# Aggregate stats for all subscriptions
curl -H "Authorization: Bearer $TOKEN" \
  http://localhost:8000/api/v1/mqtt/stats

Response:

{
  "status": "success",
  "running_count": 2,
  "subscriptions_stats": {
    "sub_abc123": {
      "messages_received": 15420,
      "bytes_received": 2458320,
      "decode_errors": 0,
      "last_message_at": "2026-02-13T10:30:15Z",
      "topics": {
        "sensors/temperature": 8500,
        "sensors/humidity": 6920
      }
    }
  }
}

Health check

curl -H "Authorization: Bearer $TOKEN" \
  http://localhost:8000/api/v1/mqtt/health

Response:

{
  "status": "healthy",
  "healthy": true,
  "running_count": 2,
  "connected_count": 2,
  "disconnected_count": 0,
  "service": "mqtt_subscriptions"
}

Prometheus metrics

Arc exposes MQTT metrics for Prometheus:

Metric	Type	Description
`arc_mqtt_messages_received_total`	Counter	Total messages received
`arc_mqtt_bytes_received_total`	Counter	Total bytes received
`arc_mqtt_decode_errors_total`	Counter	Message decode errors
`arc_mqtt_connection_status`	Gauge	Connection status (1=connected)

Querying MQTT data

Basic query

SELECT * FROM iot.temperature
ORDER BY time DESC
LIMIT 10;

Time-based aggregation

SELECT
  time_bucket(INTERVAL '5 minutes', time) as bucket,
  AVG(value) as avg_temp,
  MIN(value) as min_temp,
  MAX(value) as max_temp
FROM iot.temperature
WHERE time > NOW() - INTERVAL '1 hour'
GROUP BY bucket
ORDER BY bucket DESC;

Filter by tag

SELECT * FROM iot.sensor_data
WHERE sensor_id = 'temp-001'
  AND time > NOW() - INTERVAL '24 hours'
ORDER BY time DESC;

Troubleshooting

Connection failed

Check subscription status:

curl -H "Authorization: Bearer $TOKEN" \
  http://localhost:8000/api/v1/mqtt/subscriptions/{id}

If status is error, verify:

Broker URL is correct (tcp://, ssl://, ws://)
Broker is reachable from Arc server
Credentials are correct
TLS certificates are valid

No data appearing

Verify subscription is running (status should be "running")
Check stats for received messages
Verify topic pattern matches published topics
Check Arc logs for decode errors

Messages not decoding

Ensure messages are valid JSON or MessagePack:

# Test with simple JSON
mosquitto_pub -h localhost -t "test/data" \
  -m '{"time": 1706745600000000, "value": 42}'

Check for decode errors in stats:

curl -H "Authorization: Bearer $TOKEN" \
  http://localhost:8000/api/v1/mqtt/subscriptions/{id}/stats

Best practices

Use descriptive subscription names — Names like prod-factory-floor-sensors are easier to manage than sub1.
Separate databases by environment — Use different target databases for production, staging, and development data.
Use QoS appropriately — QoS 0 for highest throughput, QoS 1 for reliable delivery (recommended), QoS 2 for exactly-once (highest overhead).
Set reasonable reconnect intervals — Default min 1s / max 60s works well. Avoid setting min too low to prevent broker overload.
Use topic wildcards efficiently — Subscribe to specific patterns (sensors/+/temperature) rather than overly broad ones (#).
Monitor subscription health — Set up alerts on arc_mqtt_connection_status == 0 and rate(arc_mqtt_decode_errors_total[5m]) > 0.

Docker Compose example

version: '3.8'
services:
  arc:
    image: basekick/arc:latest
    ports:
      - "8000:8000"
    volumes:
      - arc-data:/data
    environment:
      ARC_MQTT_ENABLED: "true"
      ARC_AUTH_ENABLED: "true"
      ARC_ENCRYPTION_KEY: "${ARC_ENCRYPTION_KEY}"
    depends_on:
      - mosquitto

  mosquitto:
    image: eclipse-mosquitto:2
    ports:
      - "1883:1883"
    volumes:
      - ./mosquitto.conf:/mosquitto/config/mosquitto.conf

volumes:
  arc-data:

Next steps

Tiered Storage — Manage MQTT data lifecycle with hot/cold tiering
Automated Scheduling — Downsample MQTT data automatically
Audit Logging — Track MQTT subscription changes

On this page