Integrating MQTT and InfluxDB using N8N

In this tutorial, we will see how to perform the integration of MQTT and InfluxDB using N8N. This way, you will be able to store data from IoT devices that use MQTT in an InfluxDB Cloud bucket.

Figure 1 – Proposed System

General Workflow Description

The analyzed workflow is designed to receive MQTT messages from IoT devices, transform the data into the format required by InfluxDB (line protocol), and send it to InfluxDB Cloud for storage.

The workflow consists of three main nodes:

• MQTT Trigger: Receives messages from an MQTT broker.
• Code: Transforms JSON data into InfluxDB line protocol format.
• HTTP Request: Sends the transformed data to InfluxDB Cloud.

Figure 2 – Workflow

Prerequisites

Before implementing this workflow, you will need to have the following configured:

• A running N8N instance (local or in the cloud).
• Access to a cloud MQTT broker (such as EMQX, Mosquitto, HiveMQ, etc.).
• An account on InfluxDB Cloud with a configured bucket.
• IoT devices or simulators that publish data in JSON format.

Step-by-Step Implementation

Below are the steps to implement the entire workflow.

1: MQTT Broker Configuration

1. Make sure you have a configured and properly functioning MQTT broker.
2. Note the broker address, port, username, and password, as you will need them later.

2: InfluxDB Cloud Configuration

1. Create an account on InfluxDB Cloud if you do not already have one.
2. Set up a bucket to store your IoT data.
3. Generate an API token with write permissions for the bucket.
4. Note your organization URL, organization ID, bucket name, and API token.

3: Workflow Creation in N8N

MQTT Trigger Node Configuration

Follow these steps to configure the MQTT (trigger) node in N8N:

1. Create a new workflow in N8N.
2. Add an “MQTT Trigger” node to the canvas.
   • Configure MQTT broker credentials:
     • Protocol (mqtt/mqtts): In this case, we will use mqtts, which provides data encryption. However, you can use the option that fits your needs.
     • Host: Enter the hostname of your MQTT broker instance.
     • Port: The default port for unencrypted MQTT connections is 1883, while for encrypted connections it is 8883. Check this in your broker configuration.
     • Username and password: If your MQTT client uses a username and password, complete these values in the N8N node.
     • Client ID: It is recommended to leave the Client ID blank so that N8N generates one automatically each time it connects to the broker.
     • Topic: Configure the MQTT topic to subscribe to (e.g., “buenosaires/hotel01/room103”).

Code Node Configuration for Transformation

1. Add a “Code” node after the MQTT Trigger.
2. Select the “JavaScript” mode. The JavaScript code you will need depends on the format of the MQTT message payload.
   For example, if the payload is like the following, you can use the proposed JavaScript function below to build the line protocol needed to write the data to the InfluxDB bucket.

Sensor payload:

{
  "timestamp": "2025-05-30T16:30:00Z",
  "device_id": "sensor_001",
  "temperature": 25.3,
  "humidity": 60.8,
  "pressure": 1013.25,
  "location": "zona_a",
  "battery_level": 87
}

JavaScript function to format into “line protocol”, which is used by InfluxDB.

for (const item of $input.all()) {
  const payload = JSON.parse(item.json.message);
  // Timestamp conversion
  const timestampMs = Date.parse(payload.timestamp);
  const timestampNs = timestampMs * 1e6;
  // --- TAGS ---
  const tags = [];
  // Validate and sanitize each tag
  if (payload.device_id?.trim()) tags.push(`device_id=${payload.device_id.trim()}`);
  if (payload.location?.trim()) tags.push(`location=${payload.location.trim()}`);
//  if (item.json.topic?.trim()) tags.push(`topic=${item.json.topic.trim()}`);
  // --- FIELDS ---
  const fields = [];
  // Validate that fields are numeric
  if (typeof payload.temperature === 'number') fields.push(`temperature=${payload.temperature}`);
  if (typeof payload.humidity === 'number') fields.push(`humidity=${payload.humidity}`);
  if (typeof payload.pressure === 'number') fields.push(`pressure=${payload.pressure}`);
  if (typeof payload.battery_level === 'number') fields.push(`battery_level=${payload.battery_level}`);
  // --- CRITICAL VALIDATION ---
  if (tags.length === 0) throw new Error('Error: All tags are empty');
  if (fields.length === 0) throw new Error('Error: All fields are empty');
  // --- LINE BUILDING ---
  const measurement = 'iot_sensors';
  const tagsSection = tags.join(','); // E.g: "device_id=sensor_001,location=zona_a"
  const fieldsSection = fields.join(','); // E.g: "temperature=25.3,humidity=60.8"
  // Combine everything avoiding extra commas
  const line = `${measurement},${tagsSection} ${fieldsSection} ${timestampNs}`;
  return [{ json: { line } }];
}

The code performs the following operations:

• Parses the received JSON message.
• Converts the ISO8601 timestamp to nanoseconds for InfluxDB.
• Organizes the data into tags (indexed metadata) and fields (measurement values).
• Validates that tags and fields exist.
• Builds the InfluxDB line protocol in the correct format.

HTTP Request Node Configuration

Finally, configure the HTTP Request node to send the data to the InfluxDB bucket.

1. Add an “HTTP Request” node after the Code node.
2. Set the method to POST.
3. Set the URL to the InfluxDB Cloud write endpoint. Take into account the region where your InfluxDB instance is running and replace the value in the following URL: https://[REGION].aws.cloud2.influxdata.com/api/v2/write
4. Configure the query parameters (see Figure 3):
   • Your organization’s org ID.
   • Your bucket name.
   • Precision: ns (nanoseconds).
5. Configure the headers (see Figure 4):
   • Authorization: Token [your-API-token]
   • Content-Type: text/plain; charset=utf-8
   • Accept: application/json
6. In the body, use an expression to reference the line generated in the code block (see JavaScript code and Figure 4): ={{ $json.line }}

Figure 3 – URL and Query Parameters Configuration

Figure 4 – Headers and Body Configuration

Common Implementation Problems

During workflow implementation, issues such as data formatting errors, incorrect API usage, authentication errors, etc., may arise.
Let’s look at some examples that may occur for this particular workflow.

Problems with the Line Protocol Format

The most common error is receiving a 400 code with messages like “empty tag name” or “invalid line protocol”. This usually happens because:

• There is a comma between the tag and field sections (there should be a space, not a comma).
• A tag has an empty name or value.
• The timestamp format is incorrect.

The solution implemented in the sample code checks and validates all these aspects before sending the data. You can use this code as a reference to create your own formatter.

MQTT Connection Issues

If the MQTT Trigger node does not receive messages, check:

• Connectivity to the broker.
• Correct topic configuration.
• That there are no connection conflicts (leaving the Client ID blank helps).

Best Practices

For robust production environments, consider these recommendations:

• Error handling: Add IF nodes to validate data and filter malformed messages.
• Scalability: Implement write batching by grouping multiple measurements into a single request for high data volumes.
• Security: Use N8N’s credential management system and enable TLS/SSL in MQTT connections whenever possible.
• Monitoring: Set up email or webhook notifications for critical errors.

Conclusion

This workflow provides a scalable and robust solution to integrate IoT devices that use MQTT with InfluxDB Cloud. The optimized data transformation ensures reliability and efficiency in time series data ingestion, enabling real-time analysis of your IoT data.

The combination of N8N with MQTT and InfluxDB offers a flexible architecture that can be adapted to various IoT use cases, from environmental monitoring to industrial automation.